Untitled

• Shared-nothing architectures are dominant for high-performance data warehousing.
• Important architectural dimensions and methods include storage, cluster and query engine

Data Warehouse Storage

• Data layout options are column-store versus row-store.
• Column-stores read relevant data, skipping unrelated columns and are suitable for analytical workloads.
• Column stores make better use of CPU caches and leverage SIMD registers & lightweight compression.
• Column Stores make a data layout choice
• Apache Parquet and ORC are examples of column stores
• Key to storage formats is compression
• Storage formats trade I/O for CPU and are well-suited for large datasets and I/O intensive workloads.
• Column-Stores & Storage Formats work well together Ex. RLE, gzip, LZ4, etc.
• Pruning skips irrelevant data, horizontally
• Data is usually ordered so that it can maintain a sparse MinMax index.

Table Partitioning and Distribution

• Data can be spread based on a key using functions like hash, range, and list.
• Distribution is system-driven with the goal of parallelism.
• Each compute node gets a piece of the data, ensuring everyone is busy. Partitioning is user-specified to manage data lifecycle
• Data warehouses keep the last six months, loading one new day and dropping the oldest partition every night.
• It is possible to improve access pattern with partition pruning, whereby partitions P1, P3, and P4 would be dropped when querying month "May".

Query Execution

• Scalability depends how well you can make the most out of the underlying hardware.
• Vectorized execution
• Data is pipelined in batches (of few thousand rows) to save I/O, greatly improve cache efficiency. Ex. Actian Vortex, Hive, Drill, Snowflake, MySQL's Heatwave accelerator, etc.
• And/or JIT code generation
• Generate a program that executes the exact query plan, JIT compile it, and run on your data.
• Tableau/HyPer/Umbra, SingleStore, AWS Redshift, etc.

Cloud-Native Warehouses

• Design should consider good performance, scalability, elasticity, fault-tolerance, etc.
• Challenges at cloud-scale for cloud-data
• Storage abstraction from the storage format
• Distribution and data partitioning more relevant at cloud-scale.
• Data caching across the (deep) storage hierarchy influences cost/performance.
• Distributed query execution combines scale-out and scale-up.
• Distributed query optimization and global resource-aware scheduling can enhance query execution
• Think about Service form factor
• Reserved-capacity services vs. serverless instances vs. auto-scaling

Shared-Nothing Architecture

• It scales well for star-schema queires
• Very little bandwidth is required to join a small (broadcast) dimensions table and a large (partitioned) fact table.
• A shared nothing architecture has an elegant design with homogeneous nodes
• Query processor nodes have locally attached storage
• Data is horizontally partitioned across the processor nodes
• Each node is responsible for the rows on its own local disks

Worker Instance

• Each compute node has dedicated CPU, memory and locally attached disk storage
• Memory, storage, and data are partitioned among the slices.
• Hash and round-robin table partitioning / distribution are supported.
• The leader distributes data to the slices and assigns workload to them.
• The number of slices per node depends on the node size.
• The leader decides how to distribute data between the slices.
• Users can specify the distribution key to better match the query’s joins and aggregations.

Within a Slice

• Data is stored in columns/ with multi-dimensional sorting.
• It allows the collections of compression options

Fault Tolerance

• Blocks are small such as 1 MB
• Each block is replicated on a different compute node and also stored on S3.
• S3 replicates each block internally, triply.

Handling Node Failures

• In the event a node fails
• Option #1: Node 2 processes load until node 1 is restored
• Option #2: A new node is instantiated, Node 3 processes workload using data in S3 until the local disks are restored

Shared-Storage Architecture

• Separates compute and storage
• Key components are caches, metadata, transaction log, and data
• Decoupling storage and compute enables flexible scaling, scaling either layer independently.
• Storage is abundant and cheaper than compute.
• Users pay for compute needed to query working data subset.

Logic Behind Virtual Warehouses

• A dynamically created compute instances cluster with pure compute resources.
• It can be created, destroyed, and resized at any time.
• Local disk cache files are used for headers and table columns. and a local data layer
• Sizing mechanisms include # of EC2 instances and size of each instance (#cores, I/O capacity)
• Queries are mapped to a single virtual warehouse, each of which can run multiple, parallel queries.
• Virtual warehouses have access to the same shared data, without needing to copy it.

Disaggregated Compute-Storage Architectures

• Storage and compute resources can be scaled independently.
• Availability can tolerate cluster and node failures
• Heterogeneous workloads can be handled via high I/O bandwidth or heavy compute capacity.
• Compute and storage are disaggregated
• Such architectures provide multi-tenancy features and elastic data warehouses.
• Examples: Snowflake, new AWS Redshift
• Cloud storage services, e.g. AWS S3, Google Cloud Storage, Azure Blob Storage

Snowflake

• A pioneering system separating storage and compute in the cloud with 2 loosely connected, independent services
• Proprietary shared-nothing execution engine handles compute
• The compute layer is highly elastic. Storage:
• Managed Amazon S3, Azure Blob storage, or Google Cloud storage
• Dynamically cached on local storage clusters used to execute the queries

Snowflake Architecture

• Snowflake uses authentication and Access Control
• A Cloud Services Layer with Infrastructure Manager | Optimizer | Transaction Manager | Security & Metadata Storage
• A virtual warehouse with Cache
• and data storage

Snowflake's Table Storage

• Tables horizontally partitioned into large immutable files.
• Resembles pages or blocks in a traditional database system Within each file: -Values of attribute (columns) are aggregated
• Heavily compressed (gzip, RLE, etc.)
• Accelerated Query Processing:
• MinMax value of a file in column kept in catalog
• Used for running code at run-time

Disaggregated Compute-Memory-Storage Architecture

• Storage, compute, and memory can be independently scaled.
  • This allows for scheduling of resources for better utilization and supporting complex workloads.

Key features:

- A shuffle-memory layer speeds up joins and reduces I/O cost by preventing intermediate results from being written to disks.

Examples:

- BigQuery

BigQuery

• The architecture is Compute (Clusters) + Shuffle tier (Colossus DFS) + Storage and has a Dremel Query Engine The distributed memory shuffle tier supports query optimization by:
  • Reducing shuffle latency by 10x
  • Enabling 10x larger shuffles
  • Reducing resource cost by > 20%

BigQuery Shuffle Workflow

• A producer in each worker generates partitions and sends them to the shuffling layer.
• A consumer combines the partitions and performs operations locally.
• Large intermediate results can be spilled to local disks.

Stateless Shared-Storage Architectures

• Separates compute and state
• In stateful architectures, the state of in-flight transactions is stored in the compute node.
• The state is not hardened into persistent storage until the transaction commits.
• When a compute node fails, the state of non-committed transaction is lost, failing the transaction. Resilience to compute node failure and elastic data assignment must move to stateless architectures.

Stateless Compute Architectures

• Compute nodes should not hold any state information.
• Cache needs to be as close to compute as possible and can be lazily reconstructed from persistent storage, no need to be decoupled from compute
• All data, transactional logs and metadata need to be externalized
• Enables partial restart of query execution when compute nodes fail or change clusters. Ex. BigQuery uses the shuffle tier and dynamic scheduler; POLARIS

Polaris Summary

• Employs a Separation of Storage and Compute:
• Compute is done by Polaris pools with shared centralized services Metadata and Transactions.
• The stateless architecture within a pool stores data durably in remote storage, while metadata and transactional logs are offloaded to centralized services (built for high availability and performance) Features:
• Data cell abstraction processing diverse data formats and storage systems.
• Elastic query processing with separation of state and computation, flexible abstraction and fine grained organization
• Combinations of scale-up (e.g., intra-partition parallelism and vectorized processing); and scale out framework

Serverless

Serverless Computing for Queries

        Issues queries without resource provisioning and provides pay per query granularity

• Two Main Approaches:
• Serverless Databases - Cloud SQL engine and storage.
• Serverless Functions + Cloud Storage - Use Function-as-a-Service (FaaS) and cloud storage

Function-as-a-Service (FaaS)

• FaaS is serverless compute model where developers write and deploy self-contained functions - Event Driven - Response to triggers - Pay for execution time and resources Function Lifetime: Trigger - Execution - Termination. Event-Driven/ Stateless/ Autoscaling, pay-per-use.

Serverless Functions + Cloud Storage

Two challenges include Stateless functions and Stragglers that increase latency Solution:

• Use cloud storage to exchange state similar to state-separate query processing.
• Use tuned models to detect stragglers and invoke functions with duplication computation.

Serverless Functions + Cloud Storage Usage

• Query processing can be achieved using lambda functions such as the ability to involve many tasks that write the intermediate results file. Combiners can also be used to reduce the cost of the large shuffle
• As a result it shows a trade-off between the number of tasks with regards to performance vs cost

What is a Cold Start?

- It occurs if a function is invoked for the first time or after inactivity.
- The provisions are a new container or runtime environment leading to initial latency.

Shuffling Primitives

• Goal: Enable scale-out for all query plans
• Difficult to achieve robust shuffling Is a network based operator used to distribute SQL engines
• Can repartition cluster data includes -hash-based matter -range-based -random methods
• The is space is immense

Query Optimizations

Phase 1: The query undergoes Logical Optimization. Rule-Based is used for Predicate Pushdown and Redundant Join Removal Cost-Based joins Reordering based on estimated cardinalities Goal: Create an efficient equivalent to the old query Plan Phase 2: Translate to "Canonical" Distributed Plan Insert Shuffle Can be run and return with slow results to the correct Phase 3: Query undergoes Distributed Optimization -Selecting Broadcast Joins -Eliminating Redundant Shuffles -Sideways Information Passing

Query Orchestration case of Firebolt

The query plain is broken down into smaller stages with high connectedness Every stage is executed on a small set of nodes Scheduler performs topological sort on graph The stage the sent to nodes for process

Query Orchestration

It can be used with single node or multi node With source, compute, Sinks The stage is into pipelines that use morsel-driven parallelism Only shuffle needs to topology

Resource and Management Challenges

• Resource estimation
• How they overlap during execution
• Whether to query it
• Concurrency
• Cost minimal workload