PolarDB:Distributed columnar index

Horizontal sharding (sharding by tenant_id)

Assume that a business table orders exists and uses tenant_id as the sharding key.

• Single-tenant aggregation: Calculates the total and average amounts of the local orders of a specific tenant.

  SELECT SUM(amount) AS total_amount, AVG(amount) AS avg_amount
  FROM orders
  WHERE tenant_id = 1001;

• Global tenant statistics: Calculates the total and average order amounts of all tenants, grouped by tenant.

  SELECT tenant_id, SUM(amount) AS total_amount, AVG(amount) AS avg_amount
  FROM orders
  GROUP BY tenant_id;

Vertical sharding (sharding by schema)

Assume that the data of each tenant is stored in a separate schema and that the table is named orders in every schema.

• Single-schema aggregation: Aggregates the total and average order amounts of a specific schema.

  SELECT SUM(amount) AS total_amount, AVG(amount) AS avg_amount
  FROM tenant_a.orders;

• Global tenant statistics: Aggregates the total and average order amounts of all tenants by using UNION ALL.

  SELECT 'tenant_a' AS tenant, SUM(amount) AS total_amount, AVG(amount) AS avg_amount FROM tenant_a.orders
  UNION ALL
  SELECT 'tenant_b' AS tenant, SUM(amount) AS total_amount, AVG(amount) AS avg_amount FROM tenant_b.orders;

Step 1: Enable the distributed IMCI feature

You can modify the parameters in the PolarDB console or in a database session:

SET polar_cluster_enable_imci = on;
SET polar_csi.enable_anyall_to_in = on;
SET pg_hint_plan.polar_enable_distributed_hint = on;

Step 2: Create replication slots for row-to-column synchronization

Use an account with high privileges to run the following command on the primary CN to create replication slots for row-to-column synchronization on all nodes:

SELECT run_command_on_all_nodes($$ SELECT polar_csi_start_sync() $$);

Step 3: Create and shard tenant tables

-- Create the initial business table with tenant_id as part of the primary key
CREATE TABLE orders (
  order_id   bigint,
  tenant_id  int NOT NULL,
  amount     numeric(20,2),
  status     text,
  create_time timestamp,
  PRIMARY KEY (tenant_id, order_id)
);

-- Convert the table to a distributed sharded table
SELECT create_distributed_table('orders', 'tenant_id');

-- Create the tenant schema
CREATE SCHEMA tenant1;

-- Enable distributed management for the schema
SELECT polar_cluster_schema_distribute('tenant1');

-- Create the business table
SET search_path TO tenant1;

CREATE TABLE orders (
  order_id   bigint,
  tenant_id  int NOT NULL,
  amount     numeric(20,2),
  status     text,
  create_time timestamp,
  PRIMARY KEY (tenant_id, order_id)
);

Step 4: Create a columnstore index

Create a CSI columnstore index on the table:

-- Create a columnstore index on all columns
CREATE INDEX csi_orders ON orders USING CSI;

-- Create a columnstore index on specific columns only (for example, when queries involve only tenant_id and amount)
CREATE INDEX csi_orders_part ON orders1 USING CSI(tenant_id, order_id, amount);

-- Create a columnstore index without locking the table (recommended)
CREATE INDEX CONCURRENTLY csi_orders ON orders USING CSI;

Step 5: Run queries by using the columnstore index

Two parameters control whether a column-store query is run:

(Recommended) Use a hint to force a specific query to use the columnstore index:

/*+SET (polar_csi.enable_query on) SET(polar_csi.cost_threshold 0)*/
SELECT tenant_id, SUM(amount) AS total_amount, AVG(amount) AS avg_amount
FROM orders
GROUP BY tenant_id;

You can also set the parameters globally or per account. To set the parameters globally, modify them in the console. To set the parameters per account, run the following statements on the primary CN:

-- Prioritize the columnstore index for the test user
ALTER ROLE test SET polar_csi.enable_query = ON;
ALTER ROLE test SET polar_csi.cost_threshold = 0;

Step 6: Check whether a query uses the columnstore index

You can use EXPLAIN for analysis. If CSI Executor appears in the execution plan, the query uses the columnstore index:

/*+SET (polar_csi.enable_query on) SET(polar_csi.cost_threshold 0)*/
EXPLAIN (COSTS OFF) SELECT tenant_id, SUM(amount) AS total_amount, AVG(amount) AS avg_amount
FROM orders
GROUP BY tenant_id;

Example output snippet (the presence of CSI Executor indicates that the column-store query is used):

                       QUERY PLAN
---------------------------------------------------------
 Custom Scan (PolarCluster Adaptive)
   Task Count: 32
   Tasks Shown: One of 32
   ->  Task
         Node: host=127.0.0.1 port=45719 dbname=postgres
         ->  CSI Executor:
             ┌───────────────────────────┐
             │       HASH_GROUP_BY       │
             │    ────────────────────   │
             │        Aggregates:        │
             │          sum(#1)          │
             │          avg(#2)          │
             └─────────────┬─────────────┘
             ┌─────────────┴─────────────┐
             │         SEQ_SCAN          │
             │    ────────────────────   │
             │    Table: orders_102008   │
             │   Type: Sequential Scan   │
             └───────────────────────────┘