Skip to main content

Identify pushdown SQL from raw query profiles

  • Updated

Summary

Dremio allows advanced SQL query pushdown for relational database sources like Oracle, MySQL, and SQL Server, and cross-datasource joins. When pushing down SQL, Dremio rewrites the query plan to push as much computation as possible into the remote database system, resulting in scan operations marked as JDBC_SUB_SCAN in the query profile. The pushdown SQL text can be found in the "Final Physical Transformation" section of the "Planning" tab of the raw profile. 

Reported Issue

A SQL query that selects from one or more relational database (RBDMS) source tables is reduced to a simpler query plan with only a few scan operations. Some or all of these scans are labeled as JDBC_SUB_SCAN in the "Visual Profile". They are allocated only a single thread for execution and may take longer than other types of scan to run.

Overview

Two very useful features of Dremio's SQL engine are:

  1. The ability to join datasets across different data sources. For example, a table from a traditional RDBMS can by joined with a Parquet dataset in a cloud object store, like S3.
  2. Queries on tables in RDBMS data sources will include SQL statements that are submitted directly to those databases for processing. We say that some of the SQL has been "pushed down" to the data source.*

To better understand the performance of a query with joins across data sources and pushdowns, it's helpful to know how to identify which datasets are in the RDBMS sources and what SQL is being pushed down.

Relevant Versions

All versions of Dremio support trans-datasource joins and push downs are supported for RDBMS sources like Oracle, MySQL and SQL Server.

Steps to Resolve

As an example, consider the following query:

SELECT 
  cust_prod_totals.*, 
  prod_b.part_name
FROM
"Warehouse Analytics"."site_metrics"."customer_product_purchase_totals" cust_prod_totals
JOIN 
"s3_data".iceberg."product_set_b" prod_b
  ON cust_prod_totals.product_id = prod_b.product_id

This joins a view "Warehouse Analytics"."site_metrics"."customer_product_purchase_totals" with and Iceberg table in S3: "s3_data".iceberg."product_set_b".

The view encapsulates a join between 2 tables in the Oracle datasource, oracle_db. Here is the view's SQL definition:

SELECT     
  customers.CUSTOMER_ID,
  customers.FIRST_NAME,
  customers.LAST_NAME,
  sales.PRODUCT_ID,
  SUM(sales.TOTAL_PRICE) total_sales_revenue
FROM       
"oracle_db"."DREMIO_DB_USER".CUSTOMERS customers
INNER JOIN 
oracle_db.DREMIO_DB_USER.SALES_FACT sales
ON customers.CUSTOMER_ID = sales.CUSTOMER_ID
GROUP BY   
  customers.CUSTOMER_ID,
  customers.FIRST_NAME,
  customers.LAST_NAME,
  sales.product_id

This adds another join to the expanded SQL, so you would think there would be 2 joins for the query plan, but if we look at the "Visual Profile", it only shows a single join between the Iceberg table and the output of a JDBC_SUB_SCAN:

If we look to the "Final Physical Transformation" under the "Planning" tab of the raw profile, we see that the JDBC_SUB_SCAN (operator 00-10) includes the extra join as part of a SQL statement that the query planner has pushed down to the source Oracle database:

00-00    Screen : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 1000.0, cumulative cost = {11132.875 rows, 63896.56692187499 cpu, 34530.0 io, 34530.0 network, 17600.0 memory}, id = 18084
00-01      Project(Fragment=[$0], Records=[$1], Path=[$2], Metadata=[$3], Partition=[$4], FileSize=[$5], IcebergMetadata=[$6], fileschema=[$7], PartitionData=[$8], OperationType=[$9], PartitionValue=[$10], RejectedRecords=[$11], ReferencedDataFiles=[$12]) : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 1000.0, cumulative cost = {11032.875 rows, 63796.56692187499 cpu, 34530.0 io, 34530.0 network, 17600.0 memory}, id = 18083
00-02        WriterCommitter(final=[/Users/ben/dremio-releases/dremio-enterprise-26.0.1-202506021510160171-2354f09b/data/pdfs/results/17894991-a4f8-4d07-1e49-882d2ba42b00]) : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 1000.0, cumulative cost = {10032.875 rows, 63796.43692187499 cpu, 34530.0 io, 34530.0 network, 17600.0 memory}, id = 18082
00-03          Writer : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 1000.0, cumulative cost = {9032.875 rows, 62796.43692187499 cpu, 34530.0 io, 34530.0 network, 17600.0 memory}, id = 18081
00-04            Project(CUSTOMER_ID=[$0], FIRST_NAME=[$1], LAST_NAME=[$2], PRODUCT_ID=[$3], total_sales_revenue=[$4], part_name=[$5]) : rowType = RecordType(DECIMAL(38, 0) CUSTOMER_ID, VARCHAR(65536) FIRST_NAME, VARCHAR(65536) LAST_NAME, VARCHAR(65536) PRODUCT_ID, DECIMAL(38, 2) total_sales_revenue, VARCHAR(65536) part_name): rowcount = 1000.0, cumulative cost = {8032.875 rows, 61796.43692187499 cpu, 34530.0 io, 34530.0 network, 17600.0 memory}, id = 18080
00-05              Project(CUSTOMER_ID=[$0], FIRST_NAME=[$1], LAST_NAME=[$2], PRODUCT_ID=[$3], total_sales_revenue=[$4], part_name=[$6]) : rowType = RecordType(DECIMAL(38, 0) CUSTOMER_ID, VARCHAR(65536) FIRST_NAME, VARCHAR(65536) LAST_NAME, VARCHAR(65536) PRODUCT_ID, DECIMAL(38, 2) total_sales_revenue, VARCHAR(65536) part_name): rowcount = 1000.0, cumulative cost = {7032.875 rows, 61796.376921874995 cpu, 34530.0 io, 34530.0 network, 17600.0 memory}, id = 18079
00-06                HashJoin(condition=[=($3, $5)], joinType=[inner]) : rowType = RecordType(DECIMAL(38, 0) CUSTOMER_ID, VARCHAR(65536) FIRST_NAME, VARCHAR(65536) LAST_NAME, VARCHAR(65536) PRODUCT_ID, DECIMAL(38, 2) total_sales_revenue, VARCHAR(65536) product_id0, VARCHAR(65536) part_name): rowcount = 1000.0, cumulative cost = {6032.875 rows, 61796.316921875 cpu, 34530.0 io, 34530.0 network, 17600.0 memory}, id = 18072
00-08                  Project(CUSTOMER_ID=[$0], FIRST_NAME=[$1], LAST_NAME=[$2], PRODUCT_ID=[$3], total_sales_revenue=[$4]) : rowType = RecordType(DECIMAL(38, 0) CUSTOMER_ID, VARCHAR(65536) FIRST_NAME, VARCHAR(65536) LAST_NAME, VARCHAR(65536) PRODUCT_ID, DECIMAL(38, 2) total_sales_revenue): rowcount = 938.4375, cumulative cost = {2092.4375 rows, 34533.046921875 cpu, 34530.0 io, 34530.0 network, 0.0 memory}, id = 18067
00-10                    Jdbc(sql=[SELECT "CUSTOMER_ID", "FIRST_NAME", "LAST_NAME", "PRODUCT_ID", SUM("TOTAL_PRICE") "total_sales_revenue"
FROM (SELECT CAST("CUSTOMERS"."CUSTOMER_ID" AS NUMBER(38, 0)) "CUSTOMER_ID", "CUSTOMERS"."FIRST_NAME", "CUSTOMERS"."LAST_NAME", "SALES_FACT"."PRODUCT_ID", CAST("SALES_FACT"."TOTAL_PRICE" AS NUMBER(38, 2)) "TOTAL_PRICE"
FROM (SELECT "CUSTOMERS"."CUSTOMER_ID", "CUSTOMERS"."FIRST_NAME", "CUSTOMERS"."LAST_NAME"
FROM "DREMIO_DB_USER"."CUSTOMERS") "CUSTOMERS"
INNER JOIN (SELECT "SALES_FACT"."CUSTOMER_ID", "SALES_FACT"."PRODUCT_ID", "SALES_FACT"."TOTAL_PRICE"
FROM "DREMIO_DB_USER"."SALES_FACT") "SALES_FACT" ON "CUSTOMERS"."CUSTOMER_ID" = "SALES_FACT"."CUSTOMER_ID") "t"
GROUP BY "CUSTOMER_ID", "FIRST_NAME", "LAST_NAME", "PRODUCT_ID"]) : rowType = RecordType(DECIMAL(38, 0) CUSTOMER_ID, VARCHAR(65536) FIRST_NAME, VARCHAR(65536) LAST_NAME, VARCHAR(65536) PRODUCT_ID, DECIMAL(38, 2) total_sales_revenue): rowcount = 938.4375, cumulative cost = {1154.0 rows, 34533.0 cpu, 34530.0 io, 34530.0 network, 0.0 memory}, id = 17841
00-07                  Project(product_id0=[$0], part_name=[$1]) : rowType = RecordType(VARCHAR(65536) product_id0, VARCHAR(65536) part_name): rowcount = 1000.0, cumulative cost = {2002.0 rows, 1002.02 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 18071
00-09                    TableFunction(columns=[`product_id`, `part_name`], Table Function Type=[DATA_FILE_SCAN], table=[s3_data.iceberg.product_set_b]) : rowType = RecordType(VARCHAR(65536) product_id, VARCHAR(65536) part_name): rowcount = 1000.0, cumulative cost = {1002.0 rows, 1002.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 18070
00-11                      TableFunction(columns=[`splitsIdentity`, `splits`, `colIds`], Table Function Type=[SPLIT_GEN_MANIFEST_SCAN]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds): rowcount = 1.0, cumulative cost = {2.0 rows, 2.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 18069
00-12                        IcebergManifestList(table=[s3_data.iceberg.product_set_b], snapshot=[7388019168317644933], columns=[`splitsIdentity`, `splits`, `colIds`], splits=[1], metadataFileLocation=[/lakehouse-bucket-42/data/iceberg/product_set_b/metadata/v1.metadata.json], manifestContent=[DATA]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds): rowcount = 1.0, cumulative cost = {1.0 rows, 1.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 18068

Looking closer at the SQL, we can see that essentially all of the logic in the view "customer_product_purchase_totals" is being executed in Oracle, with some changes to make the SQL conform to that databases dialect:

SELECT 
  "CUSTOMER_ID", 
  "FIRST_NAME",
  "LAST_NAME", 
  "PRODUCT_ID", 
  SUM("TOTAL_PRICE") "total_sales_revenue"
FROM (
  SELECT 
    CAST("CUSTOMERS"."CUSTOMER_ID" AS NUMBER(38, 0)) "CUSTOMER_ID", 
    "CUSTOMERS"."FIRST_NAME", 
    "CUSTOMERS"."LAST_NAME", 
    "SALES_FACT"."PRODUCT_ID", 
    CAST("SALES_FACT"."TOTAL_PRICE" AS NUMBER(38, 2)) "TOTAL_PRICE"
  FROM (
    SELECT 
      "CUSTOMERS"."CUSTOMER_ID", 
      "CUSTOMERS"."FIRST_NAME", 
      "CUSTOMERS"."LAST_NAME"
    FROM "DREMIO_DB_USER"."CUSTOMERS") "CUSTOMERS"
INNER JOIN (
  SELECT 
    "SALES_FACT"."CUSTOMER_ID", 
     "SALES_FACT"."PRODUCT_ID", 
     "SALES_FACT"."TOTAL_PRICE"
  FROM "DREMIO_DB_USER"."SALES_FACT"
) "SALES_FACT" ON "CUSTOMERS"."CUSTOMER_ID" = "SALES_FACT"."CUSTOMER_ID") "t"
GROUP BY 
  "CUSTOMER_ID", 
  "FIRST_NAME", 
  "LAST_NAME", 
  "PRODUCT_ID"

Next Steps

SQL statements submitted to the the source system in this manner are made possible through the advanced relational pushdown (ARP) framework. The query planner figures out the most optimal query to push down without the need for direct input or prompting from the user. 

You can, however, explicitly write the SQL that you want submitted to a data source by an external query. External queries are only available for RDBMS sources.

Additional Resources

Dremio ARP developer guide

External Queries

 

 

*This also applies to NoSQL databases such as MongoDB and Elasticsearch, where SQL is converted into a corresponding query in their own search languages.

Was this article helpful?
0 out of 0 found this helpful
Return to top

Related articles