[Design] Add support for DELETE

[dingqiaoyi]

Background

We are making an impressive progress on the QUERY functionalities and performance improvements, but with limited support for other DMLs such as INSERT INTO，DELETE, UPDATE and MERGE INTO/UPSERT.
It is important to Velox as a unified execution engine to support both read and write workloads.

I believe adding DELETE support is a good start, hoping to lay a solid foundation to the support for other DMLs in Velox.
This document is an overall design of the support for DELETE in Velox.

---

Goals

1. [Writes at row level] Add support for DELETE at row level without the explicit dependency to the primary key constraints on the table.
2. [Transactional writes] Implement the DELETE semantic in a distributed way, providing a consistent view of the corresponding data during the execution. Note: We consider the transactional writes as a distributed commit assurance requirement here, and transactional isolation aka concurrency control, could be treat as an orthogonal requirement, will not be talked about in this topic.
3. [Avoid lock-in] We need a solution without external dependencies such as specific commit protocols, table format protocols to implement the row-level writes, etc.

---

Design

1. Writes at row level

• Optimizer & DAG/TaskManager
  • In order to support an E2E delete at row level, it is the responsibility of PrestoDB, Spark and any other systems leveraging Velox as its execution layer to generate and schedule the distributed execution plan.
  • To the best of my knowledge, the DELETE distributed plans could be generalized into some patterns described below. Different system with different levels of implementation may cover some or all of them. For example, Delta Lake 2.x is not support DELETE conditions containing subquery. The challenges are how to deal with the subqueries in the DELETE where clause and the performance considerations of the distributed delete operations.
    • P1: DELETE table FROM clause without subqueries
    • P2: DELETE table FROM clause with non-correlated or simple correlated subqueries
    • P3-P4: DELETE table FROM clause with complex correlated subqueries

P1: ... <- [delete <- filter <- scan]

P2: ... <- [delete <- filter <- join <- scan]
                <- [subquery]

P3: ... <- [delete <- filter <- ...(pull-up Execs from subquery) <- join <- scan]
                                <- [decorrelated subquery]
P4: ... <- [delete <- filter <- ...(pull-up Execs from subquery) <- join]
                               <- [scan]
                              <- [decorrelated subquery]

P4': ... <- [delete]
        <- [ filter <- ....(pull-up Execs from subquery) <- join]
                              <- [scan]
                              <- [decorrelated subquery]

P4' is an execution optimization of P4, clustering the delete operations according to storage's data partitioning by adding an extra exchange.

• Execution engine

  • The distributed execution plan is the input of Velox and will be transformed to the core:PlanNodes tree, so a core::DeleteNode is needed.
  • To implement the delete plan in Velox, we need the operator exec::Delete，which is storage agnostic by introducing a newly added Velox Connector SPI in Connector.h.
  • We are going to support delete at row level without using storage's PK indexes explicitly, so it is Velox 's duty to read out the to-delete candidate rows' identities from storage first. So, the newly added SPI may be named as UpdatableDataSource, which has the deleteRows(ids) interface and keeps the read related interfaces as well. That will be useful of optimizing executing plans in the patterns of P1-P3.

• Storage engine/Table format protocol

  • Storage engine, or the schema in the catalog to be more specifically, should take the responsibility of telling the optimizer which column is unique and could be used as row identity when generating the delete execution plan.
  • Velox is usually used in building the data processing systems adopting separation of compute and storage. Thanks to the Connector SPI abstraction, we could have this Velox design decoupling from the specific implementations of data deletion in the storage.
  • There is a trend to offload storage engine to cloud object store services with table format protocols and a set of APIs and libraries for engines to interact with them following that protocols. It will bring more work to implement those Connector SPIs. Simply forwarding delete operations will not be enough.

2. Transactional writes support

• Write commit protocol is used to achieve the consistency of output in a distributed execution engine. There is a de facto standard to break distributed commit process into 2 steps：one paralleled preparation step, and one centralized commit step.
• It is the execution engine's duty to executing the protocol. So we need a new SPI abstraction ConnectorWriteCommitProtocol. The methods in the SPI can be called from several different contexts and may be from several different components. There are two main ways to do this. Paralleled preparation methods are always called in the Write operators. The difference is how to execute the final commit step. Spark introduces a commit coordinator in charge of it into driver. Trino, PrestoDB, Flink are using another way by a modified execution DAG, adding a table commit operator to the root and setting its parallelism to 1. We will adopt the second way in this design, because of the avoid lock-in considerations.
• Commit protocols are usually defined as a part of the table format specification, such as DeltaLake, Iceberg etc. 2 dimensions can be used to evaluate the commit protocol: performance and transactional assurance. It may take some more efforts when we add a new Connector to Velox.

3. Avoid lock-in considerations

• We are introducing new Connector SPIs, and trying to implement operators based on the SPIs. It helps Velox avoid lock-in specific storages or table formats' implementations.
• DELETE is a little bit complicated than the ready-only SELECT, because it is catalog and metadata related. Some new components are needed and usually have to interact with optimizer and DAG/task manager. We are preferring implementing them as operators and reusing the data channels between them.

---

Tasks

1. [Task-1] Add DELETE support at row level without transactional support
   • Introduce the plan node and operator for delete.
   • Introduce a Connector SPI, UpdatableDataSource.
   • They will be tested by the integrated test cases in the scope of one Velox task.
2. [Task-2] Add support for write commit protocol execution in Velox without external dependencies
   • Remove Presto specific commit protocol implementations from TableWriter operator.
   • Introduce a Connector SPI, WriteCommitProtocol, a generalized commit protocol interface for commit aware operators such as TableWriter, Delete, Update, etc.
   • Introduce a TableFinish operator in Velox. It is a necessary when integrating Velox with Spark and Flink. It may also be a preference to Presto-on-Velox, but some changes are needed in the current optimizer and stage scheduler in PrestoDB.
   • Integrated testing is needed in the scope of Velox using mock Connector SPIs.
3. [Task-3] Add support for transactional INSERT INTO to Hive connector.
   • Refactoring of TableWriter operator is needed.
4. [Task-4] Refactor HIVE connector to support DELETE
   • Implements the SPIs of UpdatabaleDataSource and WriteCommitPrtotocol in Velox.
   • ASFAIK, Hive using the Base+Delta way of supporting DELETE at row-level, as a result some refactoring are need in the HiveDataSource too.
   • At this time, we can test the DELETE support in VELOX using Hive connector.
   • To make it into a more production-ready functionality, compaction support in the form of VACUUM may be needed too. While it can be considered as an orthogonal task.
5. [Task-5] Add E2E DELETE support for Presto-on-Velox / Spark-on-Velox.
   • Some protocol transformation works are needed in 'PrestoDB/presto-native-execution/presto_cpp' for example.
6. [Task-6] Add support for DELETE to more Connectors such as delta lake, iceberg and hudi.

---

Plan

• There's a task dependency graph below and some milestones are set.
• Sub tasks may be needed when we dig deeper into a specific task, for example Task 4: Refactor HIVE connector to support DELETE
• There is already a PR Add DELETE support at row level without transactional support #2382 for Task-1, I will do some refactoring to address the comments; and another PR for Task-2 is going to be created.

Task-1
( Milestone1: Velox supports DELETE at-row-level w/o transactional guarantees)
Task-2 <- Task-3
( Milestone2: Velox supports transactional writes)
        ( Milestone3: Velox supports E2E Hive insert into w/ transactional gurrantee)
                <- Task 4
                ( Milestone4: Velox supports E2E Hive DELETE w/ transactional gurrantee)
                <- Task 5
                ( Milestone5: Presto-on-Velox supports E2E DELETE at row level w/ transactional gurrantee)
                <- Task 6

[dingqiaoyi]

@mbasmanova @majetideepak

[mbasmanova]

@dingqiaoyi Thank you for a detailed write up. I do not understand this fully, hence, some questions.

The design above lists a few DELETE query patterns. Would it be possible to provide examples of queries for each pattern? This will help understand what "simple" and "complex" correlated subqueries mean. Also, would you clarify that does "(pull-up Execs from subquery)" mean?

P1: DELETE table FROM clause without subqueries
P2: DELETE table FROM clause with non-correlated or simple correlated subqueries
P3-P4: DELETE table FROM clause with complex correlated subqueries

The distributed execution plan is the input of Velox

Just to clarify, Velox takes a single fragment as input, not the whole distributed plan.

We are going to support delete at row level without using storage's PK indexes explicitly, so it is Velox 's duty to read out the to-delete candidate rows' identities from storage first.

This requires scan to be collocated with the join and therefore limits the set of supported queries. These queries would need to either broadcast join build side (which implies that join build side is relatively small) or make sure that both probe and build side are bucketed on the join key. I wonder what do you think about these limitations and whether it is worth calling them out explicitly in the design proposal above.

There is a trend to offload storage engine to cloud object store services with table format protocols and a set of APIs and libraries for engines to interact with them following that protocols.

Would you give some examples to help understand this statement?

DELETE is a little bit complicated than the ready-only SELECT, because it is catalog and metadata related. Some new components are needed and usually have to interact with optimizer and DAG/task manager. We are preferring implementing them as operators and reusing the data channels between them.

Would you clarify a bit what do you mean by "usually have to interact with optimizer and DAG/task manager" and how you are thinking of implementing this logic "as operators and reusing the data channels". Some examples would be helpful.

[gggrace14]

Hi @dingqiaoyi ,

We're also looking at commit protocol for DMLs. And I like the idea of abstracting into an SPI WriteCommitProtocol.

It'd be nice if you could include more specifics on how DELETE works on the level of TableWriter.

I'm not sure if it's necessary to introduce an additional TableFinish operator to Velox execution. If it's intended for distributed commit, can we do it in TableWriter and save an additional operator? As you might know, Presto today has the implementation of TableFinish commit operator, which is scheduled to run on the centralized coordinator side, w/ parallelism set to 1 as you pointed out, and primary performs Metastore commit that has to run on the coordinator. It is too much to add Metastore support to Velox.

Another aspect that I'd like to have your thought is, should commit protocol be at the level of connector or catalog? Ex., Hive connector could operate on HDFS or local file system. We might want to have different commit protocol for the two.