Distributed columnar index

Horizontal sharding (by tenant_id)

Consider a business table named orders that uses tenant_id as the sharding key.

• Aggregate data for a single tenant: Calculate the total and average order amount for a specific tenant.

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

• Aggregate data across all tenants: Calculate the total and average order amount for 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 (by schema)

Consider that each tenant's data is stored in a separate schema, and the table name is orders in all schemas.

• Aggregate data for a single schema: Calculate the total and average order amount within a specific schema.

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

• Aggregate data across all tenants: Calculate the total and average order amount for 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 distributed IMCI

You can modify the following parameters in the PolarDB console or within 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

Use an account with high privileges to run the following command on the primary CN. This creates a replication slot on all nodes for synchronization between the row-store and column-store.

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 it to a distributed sharded table
SELECT create_distributed_table('orders', 'tenant_id');

-- Create a schema for the tenant
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 columnar index

Create a columnar index (CSI) on the table:

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

-- Create a columnar index on specific columns (e.g., for queries on tenant_id, order_id, and amount)
CREATE INDEX csi_orders_part ON orders1 USING CSI(tenant_id, order_id, amount);

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

Step 5: Execute queries with columnar index

Two parameters control whether a query uses the columnar index:

We recommend using a hint to force a specific query to use the columnar 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 these parameters globally or per user account. To set them globally, modify the parameters in the PolarDB console. To set them for a user, run the following statements on the primary CN:

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

Step 6: Verify columnar index usage

You can use EXPLAIN for analysis. If CSI Executor appears in the execution plan, it indicates that the columnar index is used successfully:

/*+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 CSI Executor indicates that the query used the column-store engine):

                       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   │
             └───────────────────────────┘