Sharding APIs for Oracle UCP

A shard-aware application gets a connection to a given shard by specifying the sharding key using the enhanced sharding API calls createShardingKeyBuilder and createConnectionBuilder.

At a high-level, the following steps have to be followed in making an application work with a Sharded Database:

1. Update the URL to reflect the Shard Directors and Global Service.

2. Set the pool parameters at the pool level and the shard level:

   • Initial number of connections to be created when UCP is started using setInitialPoolSize

   • Minimum number of connections maintained by pool at runtime using setMinPoolSize

   • UCP property to set maximum number of connections allowed on connection pool using setMaxPoolSize

   • Set max connections per shard with setMaxConnectionsPerShard

3. Build a sharding key object with createShardingKeyBuilder.

4. Establish a connection using createConnectionBuilder.

5. Execute transactions within the scope of the given shard.

ODP.NET Sharding APIs

A shard-aware application gets a connection to a given shard by specifying the sharding key and super sharding key using ODP.NET APIs, such as the SetShardingKey(OracleShardingKey shardingKey, OracleShardingKey superShardingKey) instance method on the OracleConnection class.

At a high level, the following steps are necessary for a .NET application to work with a Sharded Database:

1. Use ODP.NET, Unmanaged Driver.

   Sharding is supported with or without ODP.NET connection pooling. Each pool can maintain connections to different shards of the sharded database.

2. Use an OracleShardingKey class to set the sharding key and another instance for the super sharding key.

3.  Invoke the OracleConnection.SetShardingKey() method prior to calling OracleConnection.Open() so that ODP.NET can return a connection with the specified sharding key and super sharding key.

   These keys must be set while the OracleConnection is in a Closed state, otherwise an exception is thrown.

Connecting to the Multi-Shard Coordinator

To perform multi-shard queries, connect to the multi-shard coordinator using the GDS$CATALOG service on the shard catalog database:

sqlplus app_schema/app_schema@shardcatvm:1521/GDS\$CATALOG.oradbcloud

Multi-Shard Coordinator High Availability

The availability of the multi-shard coordinator impacts proxy-routing based workloads, so it is highly recommended that the coordinator be protected with Data Guard in Maximum Availability protection mode (zero data loss failover) with fast-start failover enabled. The coordinator may optionally be Oracle RAC-enabled for additional availability and scalability.

To improve the scalability and availability of multi-shard query workloads, Oracle Active Data Guard standby shard catalog databases in read-only mode can act as multi-shard query coordinators. For each active replica of the catalog database, a special coordinator service, GDS$COORDINATOR.cloud_name (where cloud_name is the value specified for the configname parameter in the GDSCTL command, create shardcatalog, and is oradbcloud by default) is running and registered on all shard directors. Clients can connect to this service on any of the replicas and perform multi-shard queries, allowing shard directors to distribute the multi-shard query workload with respect to runtime load balancing and decrease the load on in the primary shard catalog, which is the central component of the Oracle Sharding framework. Additionally, if the database’s region is set, and the client specifies the region in the connection string, a shard director routes a connection with respect to regional affinity.

Availability of the multi-shard query coordinator has zero impact on workloads using direct routing.

Multi-Shard Query Coordinator Database Sizing

The shard catalog and multi-shard query coordinator host the following key information:

• Metadata of the sharded database topology

• Schema of the sharded application

• Master copies of the duplicated tables

The size of the metadata and the schema is nominal; however, the number of duplicated tables and the space they occupy should be planned for when sizing the multi-shard query coordinator.

In addition to the above, the multi-shard query coordinator should also be sized to handle proxy routing, which can be CPU, I/O, and memory intensive based on the SQL queries and the amount of data being processed.

Restrictions

The following are restrictions on the use of query constructs in Oracle Sharding:

• Restrictions on DMLs are given with examples in the Transactions section below.

• CONNECT BY queries are not supported.

• User-defined PL/SQL is allowed in multi-shard queries only in the SELECT clause. If it is specified in the WHERE clause then an error is thrown.

  XLATE, XML query and Object type column are not supported.

• IAS on sharded table on coordinator is not supported.

• MODEL clause is not supported.

Query Involves Only Duplicated Tables

For queries that involve only duplicated tables there are no restrictions on the query shape. The query is executed on the coordinator.

Query Involves Only Sharded Tables

For a single table query, the query can have an equality filter on the sharding key that qualifies a shard. For join queries, all of the tables should be joined using equality on the sharding key. The following are some examples of queries involving sharded tables.

Inner join where equi-join is only on sharding key.

SELECT … FROM s1 INNER JOIN s2 ON s1.sk=s2.sk 
WHERE any_filter(s1) AND any_filter(s2)

Left outer join only on sharding key.

SELECT … FROM s1 LEFT OUTER JOIN s2 ON s1.sk=s2.sk

Right outer join, same as left outer join.

SELECT … FROM s1 RIGHT OUTER JOIN s2 ON s1.sk=s2.sk

Full outer join only on the sharding key, but only if Native is valid which requires equi-join.

SELECT … FROM s1 FULL OUTER JOIN s2 ON s1.sk=s2.sk
WHERE any_filter(s1) AND any_filter(s2)

Query Involves Sharded and Duplicated Tables

A query involving both sharded and duplicated tables can be either a single-shard or multi-shard query, based on the predicates on the sharding key. The only difference is that the query will contain a non sharded table which is duplicated on each shard.

Joins between a sharded table and a duplicated table can be on any column using any comparison operator (= < > <= >=) or even arbitrary join expression. The following are examples of join patterns.

Inner join

SELECT … FROM s1 INNER JOIN r1 ON any_join_condition(s1,r1) 
WHERE any_filter(s1) AND any_filter(r1)

Left/Right outer join

Where sharded table is the first table in LEFT OUTER JOIN:

SELECT … FROM s1 LEFT OUTER JOIN r1 ON any_join_condition(s1,r1) 
WHERE any_filter(s1) AND any_filter(r1)

SELECT … FROM r1 LEFT OUTER JOIN s1 ON any_join_condition(s1,s2) 
AND any_filter(r1) AND filter_one_shard(s1)

Where sharded table is second table in RIGHT OUTER JOIN:

SELECT … FROM r1 RIGHT OUTER JOIN s1 ON any_join_condition(s1,r1) 
WHERE any_filter(s1) AND any_filter(r1)

SELECT … FROM s1 RIGHT OUTER JOIN r1 ON any_join_condition(s1,s2) 
AND filter_one_shard(s1) AND any_filter(r1)

Full outer join

SELECT … FROM s1 FULL OUTER JOIN r1 ON s1.sk=s2.sk
WHERE any_filter(s1) AND any_filter(s2)

Semi-join (EXISTS)

SELECT … FROM s1 EXISTS 
(SELECT 1 FROM r1 WHERE r1.anykey=s1.anykey)

SELECT … FROM r1 EXISTS 
(SELECT 1 FROM s1 WHERE r1.anykey=s1.anykey and filter_one_shard(s1))

Anti-join (NOT EXISTS)

SELECT … FROM s1 NOT EXISTS 
(SELECT 1 FROM r1 WHERE r1.anykey=s1.anykey)	

Left/Right outer join

Where the duplicated table is the first table in LEFT OUTER JOIN, or the sharded table is first and it maps to a single shard based on filter predicate on sharding key:

SELECT … FROM r1 LEFT OUTER JOIN s1 ON any_join_condition(s1,s2) 
AND any_filter(r1) AND any_filter(s1)

Where the duplicated table is the second table in RIGHT OUTER JOIN, or the sharded table is second and it maps to a single shard based on filter predicate on sharding key:

SELECT … FROM s1 RIGHT OUTER JOIN r1 ON any_join_condition(s1,s2) 
AND any_filter (s1) AND any_filter(r1)

Full Outer Join

Sharded table requiring access to multiple shards:

SELECT … FROM s1 FULL OUTER JOIN r1 ON s1.non_sk=s2.non_sk
WHERE any_filter(s1) AND any_filter(s2)

Semi-join (EXISTS)

Sharded table is in a subquery that requires access of multiple shards:

SELECT … FROM r1 EXISTS 
(SELECT 1 FROM s1 WHERE r1.anykey=s1.anykey)

Anti-join (NOT EXISTS)

Sharded table is in the sub-query:

SELECT … FROM r1 NOT EXISTS 
(SELECT 1 FROM s1 WHERE r1.anykey=s1.anykey

Aggregate Functions

The following aggregations are supported by proxy routing:

• COUNT

• SUM

• MIN

• MAX

• AVG

Multi-shard Queries and Global CR

A multi-shard query must maintain global read consistency (CR) by issuing the query at the highest common SCN across all the shards. See Specifying Consistency Levels in a Multi-Shard Query for information about how to set consistency levels.

Transactions

A DML statement that affects only one shard is supported. For example,

update S1 set col = … where sk = <constant>;

A DML statement that affects more than one shard is not supported. For example,

update S1 set col = …;

Within a transaction, multiple single shard DMLs can be performed on different shards. For example,

insert into S1 values (…);
Update S1 set col = … where sk = constant;
Delete S1 where sk = constant;
Commit;

For multi-shard DML, the coordinator uses database link, starts a distributed transaction and performs two phase commit to guarantee the consistency of the distributed transaction. In the case of an in-doubt transaction, the database administrator has to recover it manually.

Passing Hints in Multi-Shard Queries

Any hint specified in the original query on the coordinator is propagated to the shards.

Tracing and Troubleshooting Slow Running Multi-Shard Queries

Set the trace event shard_sql on the coordinator to trace the query rewrite and shard pruning. One of the common performance issues observed is when the GROUP BY is not pushed to the shards because of certain limitations of the sharding. Check if all of the possible operations are pushed to the shards and the coordinator has minimal work to consolidate the results from shards.

SQL segment interfaces for dbms_xplan.display_cursor()

Two interfaces display the plan of a SQL segment executed on shards. The interfaces take shard IDs as the argument to display the plans from the specified shards . The ALL_SHARDS format displays the plans from all of the shards.

To print all the plans from shards use the format ALL_SHARDS as shown here.

select * from table(dbms_xplan.display_cursor(sql_id=>:sqlid,
                                              cursor_child_no=>:childno,
                                              format=>'BASIC +ALL_SHARDS‘,
                                              shard_ids=>shard_ids))

To print selective plans from the shards, pass shard IDs in the display_cursor() function. For plans from multiple shards pass an array of numbers containing shard IDs of interest in the shard_ids parameter as shown here.

select * from table(dbms_xplan.display_cursor(sql_id=>:sqlid, 
                                               cursor_child_no=>:childno,
                                               format=>'BASIC',
                                               shard_ids=>ids))

To return a plan from one shard pass the shard ID directly to the shard_id parameter, as shown here.

select * from table(dbms_xplan.display_cursor(sql_id=>:sqlid,
                                              cursor_child_no=>:childno,
                                              format=>'BASIC',
                                              shard_id=>1))

[V/X]$SHARD_SQL

[V/X]$SHARD_SQL uniquely maps a shard SQL fragment of a multi-shard query to the target shard database. This view is relevant only for the shard coordinator database to store a list of shards accessed for each shard SQL fragment for a given multi-shard query. Every execution of a multi-shard query can execute a shard SQL fragment on different set of shards, so every execution updates the shard IDs. This view maintains the SQL ID of a shard SQL fragment for each REMOTE node and the SHARD IDs on which the shard SQL fragment was executed.

Name                                      Null?    Type
----------------------------------------- -------- ----------------------------
 SQL_ID                                            VARCHAR2(13)
 CHILD_NUMBER                                      NUMBER
 NODE_ID                                           NUMBER
 SHARD_SQL_ID                                      VARCHAR2(13)
 SHARD_ID                                          NUMBER
 SHARD_CHILD_NUMBER                                NUMBER

• SQL_ID – SQL ID of a multi-shard query on coordinator

• CHILD_NUMBER – cursor child number of a multi-shard query on coordinator

• NODE_ID – ID of REMOTE node for a shard SQL fragment of a multi-shard query

• SHARD_SQL_ID – SQL ID of the shard SQL fragment for given remote NODE ID

• SHARD_ID – IDs of shards where the shard SQL fragment was executed

• SHARD _CHILD_NUMBER– cursor child number of a shard SQL fragment on a shard (default 0)

The following is an example of a multi-shard query on the sharded database and the execution plan.

SQL> select count(*) from departments a where exists (select distinct department_id
 from departments b where b.department_id=60);
------------------------------------------------
| Id  | Operation          | Name              |
------------------------------------------------
|   0 | SELECT STATEMENT   |                   |
|   1 |  SORT AGGREGATE    |                   |
|   2 |   FILTER           |                   |
|   3 |    VIEW            | VW_SHARD_377C5901 |
|   4 |     SHARD ITERATOR |                   |
|   5 |      REMOTE        |                   |
|   6 |    VIEW            | VW_SHARD_EEC581E4 |
|   7 |     SHARD ITERATOR |                   |
|   8 |      REMOTE        |                   |
------------------------------------------------

A query of SQL_ID on the V$SHARD_SQL view.

SQL> Select * from v$shard_sql where SQL_ID = ‘1m024z033271u’;
SQL_ID        NODE_ID   SHARD_SQL_ID  SHARD_ID
------------- -------  -------------- --------
1m024z033271u       5   5z386yz9suujt        1
1m024z033271u       5   5z386yz9suujt       11 
1m024z033271u       5   5z386yz9suujt       21 
1m024z033271u       8   8f50ctj1a2tbs 	    11