Idempotent Updates of Partitions

[haydenflinner]

I have a couple of use-cases for idempotent updates, probably to be solved differently.

Dimensional tables

For dimensional tables, like CREATE TABLE IF NOT EXISTS colors ( color_id INTEGER NOT NULL, color_name STRING NOT NULL );
How should I update this table daily?

1. In postgres I can INSERT IGNORE DUPLICATE.
2. With Spark/Iceberg, I can write a dataframe in overwrite mode so it replaces what's there.
3. Similarly in databend I can INSERT OVERWRITE. (Note, this isn't natively supported in sqlalchemy mysql flavor, I tried to get it to work so that I could use pandas df.write_sql, but to no avail.)
   This is good enough for these small tables, although I wonder if there will be noticeable overhead from doing this daily, or if I should instead pull the table, check if it's different from what i'm about to push, then INSERT normally if new values have been added.

Fact tables

Now, for fact tables, like: CREATE TABLE IF NOT EXISTS clicks ( event_id STRING PRIMARY KEY, date STRING, metadata STRING, ts TIMESTAMP ); with 1GB/day inserted.
If I am pulling into databend from some other data source, and realize my copy job failed halfway through, or there was a bug in my copying code that is now fixed, it's not clear to me how I can idempotently re-run the upload job. If I run the job again, there will be duplicate entries, more than one row returned with same event_id. One thing I can do is DELETE FROM clicks WHERE date = '20210401';, and then reupload all data for that date. I guess this will do a 'copy-on-write' style thing that is similar to iceberg's, and might be performant enough since these reruns will only be ad-hoc / rarely run. But in iceberg this is much easier because I don't have to explicitly delete the data myself, I can just partition the table by date, and then use overwritePartitions to be sure that only the most recent job that uploaded has any data in the clicks table for this date. Is there any easier way than DELETE WHERE date='xyz', INSERT?

Transactions

In both of these cases, although it's more likely with the fact-table historical updates, I feel like it's possible that someone reads from my databend between my DELETE and my INSERT. Is there any support for multi-statement transactions? I guess I don't need cross-table transactions, and I can get away without them if the DELETE is atomic and the INSERT is atomic. I can cross-reference on the read side to make sure all data I require is present; but if e.g. a read might return half of the data I inserted from a file, if an insert is not atomic, then I'd have a problem.

[dantengsky]

Dimensional tables

....This is good enough for these small tables, although I wonder if there will be noticeable overhead from doing this daily, or if I should instead pull the table, check if it's different from what i'm about to push, then INSERT normally if new values have been added.

INSERT OVERWRITE is preferred in this case.

Fact tables

Now, for fact tables, like: CREATE TABLE IF NOT EXISTS clicks ( event_id STRING PRIMARY KEY, date STRING, metadata STRING, ts TIMESTAMP ); with 1GB/day inserted.
If I am pulling into databend from some other data source, and realize my copy job failed halfway through,

is COPY INTO <TABLE> being used here?

there is a built-in "idempotent copy" mechanism there, which seems not been put into the doc :( though.

e.g. it is safe to safe to re-run the following stmt (by default, within 7 days)

COPY INTO books
  FROM 's3://databend-bohu/data/'
  credentials=(aws_key_id='<your-access-key-id>' aws_secret_key='<your-secret-access-key>' [aws_token='<your-aws-temporary-access-token>'])
  pattern ='.*[.]csv'
  file_format = (type = CSV field_delimiter = ','  record_delimiter = '\n' skip_header = 0);

databend meta layer tracks the data files that have been ingested, duplicated data files will be detected and prevented from being processed again.

the default 7 days period of the deduplication protection could be adjusted by tuning the load_file_metadata_expire_hours

or there was a bug in my copying code that is now fixed, it's not clear to me how I can idempotently re-run the upload job. If I run the job again, there will be duplicate entries, more than one row returned with same event_id.
One thing I can do is DELETE FROM clicks WHERE date = '20210401';, and then reupload all data for that date. I guess this will do a 'copy-on-write' style thing that is similar to iceberg's, and might be performant enough since these reruns will only be ad-hoc / rarely run.

based on the volume and the "pattern" of daily ingestion and the "pattern", DELETE FROM clicks WHERE date = '20210401' should be performant enough, most of the underlying operations are likely to be table-meta-only operations.

But in iceberg this is much easier because I don't have to explicitly delete the data myself, I can just partition the table by date, and then use overwritePartitions to be sure that only the most recent job that uploaded has any data in the clicks table for this date. Is there any easier way than DELETE WHERE date='xyz', INSERT?

there is a alter table flashback stmt, which can "revert" the state of the table to the given point(snapshot or timestamp), seems to be handy as well. but, this stmt may lose some of the table history, if it does matter, delete and re-upload the correct delta data is preferred.

Transactions

In both of these cases, although it's more likely with the fact-table historical updates, I feel like it's possible that someone reads from my databend between my DELETE and my INSERT. Is there any support for multi-statement transactions? I guess I don't need cross-table transactions, and I can get away without them if the DELETE is atomic and the INSERT is atomic. I can cross-reference on the read side to make sure all data I require is present; but if e.g. a read might return half of the data I inserted from a file, if an insert is not atomic, then I'd have a problem.

DELETE and INSERT is atomic

[haydenflinner]

e.g. it is safe to safe to re-run the following stmt (by default, within 7 days)

I find this feature pretty surprising, but I guess it's handy for someone. I am using INSERT INTO, but with the streaming local-file HTTP endpoint, because it felt easier than pushing my file to S3 / file wasn't already in S3.

DELETE and INSERT is atomic

Awesome! Are these guarantees documented anywhere? I am also curious about a general overview of the file format used, I poked around but the doc overview seems to only mention the indexing scheme, none of this metadata. Also, that page says Each Parquet file is sorted by the primary key before..., but there are no primary keys. So not sure how much I can trust it.

Thank you for your answer!

[dantengsky]

you are welcome!

I find this feature pretty surprising, but I guess it's handy for someone. I am using INSERT INTO, but with the streaming local-file HTTP endpoint, because it felt easier than pushing my file to S3 / file wasn't already in S3.

"Internal Stage" may help in this scenario :

https://databend.rs/doc/load-data/stage#step-1-create-stage-object

also, there is a copy option PURGE for copy-into statement, when set to true, successfully loaded data files will be removed from stage automatically.

Awesome! Are these guarantees documented anywhere?
I am also curious about a general overview of the file format used, I poked around but the doc overview seems to only mention the indexing scheme, none of this metadata. Also, that page says Each Parquet file is sorted by the primary key before..., but there are no primary keys. So not sure how much I can trust it.

documents are out of sync, sorry for that. by default table data is stored as parquet files ("tailored" but standard-compatible), new storage format strawboat is under developing/researching.

like snowflake, the uniqueness constraint of the primary key is not enforced.