Design of Supporting Shuffling in Input Port Materialization Reader Threads #3392

Background

Based on our offline discussions, we believe using Input Port Materialization Threads to Replace Cache Read operators makes the system cleaner mainly on the following two things:

• It is symmetric to the design of using output port materialization writer threads (Use Output Ports of an Operator to Write Storage #3295)
• It enables the scheduler to make its input plan immutable since we are no longer adding more operators during scheduling.

Supporting Multiple Workers (Shuffling) for Input Port Materialization Reader Threads

A main complexity of this new design is where to put the shuffling logic in a cache read operator, since shuffling is necessary when there are multiple workers for one operator/input port.

We initially wanted to utilize Apache Iceberg's hidden partitioning feature, but further research and discussion revealed it is not currently possible to use Iceberg's partitioning for our shuffling (or any data system's shuffling).
Here is our conclusion about Iceberg Partitioning:

• Iceberg is used only as source and sink (result) operators in other data engines like Spark/Flink.
• The internal materializations during a Spark job execution currently have nothing to do with Iceberg:
  • In the early years of Spark this materialization is stored as local files of each worker.
  • Later Spark introduced external shuffle service to address issues of that previous simple design.
• Iceberg APIs are only concerned with partitioning and currently only supports hash / range based partitioning (because these two are deterministic).
• If an Iceberg source/sink in a Spark job has a different partitioning scheme from the downstream/upstream operator, a shuffling Spark operator will be introduced (similar to the cache read op we are trying to remove).
• Shuffling differs from partitioning, and for the current task of removing cache source, what we need to preserve is the shuffling logic of a cache source op (includes round-robin shuffling, broadcast shuffling, etc.), which is not achievable by directly using Iceberg Partitioning API.
• Texera/Amber differs from existing data systems in this sense because we rely on Iceberg to store both results and intermediate materializations.

Conclusion

Our current simple design is to move the shuffling logic of a cache source op to the corresponding input port materialization reader thread

• Cons: This will introduce additional overhead because of duplicate read by multiple workers.
• Pros: Shuffling information on an input port will be part of the resource config of that port, which can be utilized by cost estimator and resource allocator because shuffling is an expensive resource

Future Project

In the future, we may want to further optimize the shuffling on an input port, possibly by designing a partitioned materialization system (on top of Iceberg) that considers shuffling. This will be a separate project.