BlazingDB · gcca · Apr 9, 2021 · Apr 9, 2021 · Apr 9, 2021 · Apr 11, 2021
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -23,6 +23,7 @@
 - #1396 Create tables from other RDBMS
 - #1427 Support for CONCAT alias operator
 - #1424 Add get physical plan with explain
+- #1472 Implement predicate pushdown for data providers
 
 ## Improvements
 - #1325 Refactored CacheMachine.h and CacheMachine.cpp 

diff --git a/README.md b/README.md
@@ -70,7 +70,7 @@ bc.sql('SELECT passenger_count, trip_distance FROM taxi LIMIT 2')
 ## Documentation
 You can find our full documentation at [docs.blazingdb.com](https://docs.blazingdb.com/docs).
 
-# Prerequisites 
+# Prerequisites
 * [Anaconda or Miniconda](https://docs.conda.io/projects/conda/en/latest/user-guide/install/linux.html) installed
 * OS Support
   * Ubuntu 16.04/18.04 LTS
@@ -96,7 +96,7 @@ Where $CUDA_VERSION is 10.1, 10.2 or 11.0  and $PYTHON_VERSION is 3.7 or 3.8
 *For example for CUDA 10.1 and Python 3.7:*
 ```bash
 conda install -c blazingsql -c rapidsai -c nvidia -c conda-forge -c defaults blazingsql python=3.7 cudatoolkit=10.1
-``` 
+```
 
 ## Nightly Version
 ```bash

diff --git a/engine/src/cython/engine.cpp b/engine/src/cython/engine.cpp
@@ -25,6 +25,11 @@
 #include "../io/data_provider/sql/MySQLDataProvider.h"
 #endif
 
+#ifdef POSTGRESQL_SUPPORT
+#include "../io/data_parser/sql/PostgreSQLParser.h"
+#include "../io/data_provider/sql/PostgreSQLDataProvider.h"
+#endif
+
 #ifdef SQLITE_SUPPORT
 #include "../io/data_parser/sql/SQLiteParser.h"
 #include "../io/data_provider/sql/SQLiteDataProvider.h"
@@ -88,12 +93,23 @@ std::pair<std::vector<ral::io::data_loader>, std::vector<ral::io::Schema>> get_l
 #else
       throw std::runtime_error("ERROR: This BlazingSQL version doesn't support MySQL integration");
 #endif
-    } else if(fileType == ral::io::DataType::SQLITE) {
+    } else if(fileType == ral::io::DataType::POSTGRESQL) {
+#ifdef POSTGRESQL_SUPPORT
+		parser = std::make_shared<ral::io::postgresql_parser>();
+    auto sql = ral::io::getSqlInfo(args_map);
+    provider = std::make_shared<ral::io::postgresql_data_provider>(sql, total_number_of_nodes, self_node_idx);
+	isSqlProvider = true;
+#else
+      throw std::runtime_error("ERROR: This BlazingSQL version doesn't support PostgreSQL integration");
+#endif
+	} else if(fileType == ral::io::DataType::SQLITE) {
 #ifdef SQLITE_SUPPORT
   		parser = std::make_shared<ral::io::sqlite_parser>();
       auto sql = ral::io::getSqlInfo(args_map);
       provider = std::make_shared<ral::io::sqlite_data_provider>(sql, total_number_of_nodes, self_node_idx);
       isSqlProvider = true;
+#else
+      throw std::runtime_error("ERROR: This BlazingSQL version doesn't support SQLite integration");
 #endif
     }
 
@@ -184,7 +200,7 @@ std::shared_ptr<ral::cache::graph> runGenerateGraph(uint32_t masterIndex,
 {
   using blazingdb::manager::Context;
   using blazingdb::transport::Node;
-  
+
   auto& communicationData = ral::communication::CommunicationData::getInstance();
 
   std::vector<Node> contextNodes;

diff --git a/engine/src/cython/io.cpp b/engine/src/cython/io.cpp
@@ -17,6 +17,11 @@
 #include "../io/data_provider/sql/MySQLDataProvider.h"
 #endif
 
+#ifdef POSTGRESQL_SUPPORT
+#include "../io/data_parser/sql/PostgreSQLParser.h"
+#include "../io/data_provider/sql/PostgreSQLDataProvider.h"
+#endif
+
 #ifdef SQLITE_SUPPORT
 #include "../io/data_parser/sql/SQLiteParser.h"
 #include "../io/data_provider/sql/SQLiteDataProvider.h"
@@ -71,6 +76,17 @@ TableSchema parseSchema(std::vector<std::string> files,
 		parser = std::make_shared<ral::io::mysql_parser>();
     auto sql = ral::io::getSqlInfo(args_map);
     provider = std::make_shared<ral::io::mysql_data_provider>(sql, 0, 0);
+#else
+      throw std::runtime_error("ERROR: This BlazingSQL version doesn't support MySQL integration");
+#endif
+    isSqlProvider = true;
+  } else if(fileType == ral::io::DataType::POSTGRESQL) {
+#ifdef POSTGRESQL_SUPPORT
+		parser = std::make_shared<ral::io::postgresql_parser>();
+    auto sql = ral::io::getSqlInfo(args_map);
+    provider = std::make_shared<ral::io::postgresql_data_provider>(sql, 0, 0);
+#else
+      throw std::runtime_error("ERROR: This BlazingSQL version doesn't support PostgreSQL integration");
 #endif
     isSqlProvider = true;
   } else if(fileType == ral::io::DataType::SQLITE) {
@@ -79,6 +95,8 @@ TableSchema parseSchema(std::vector<std::string> files,
     auto sql = ral::io::getSqlInfo(args_map);
     provider = std::make_shared<ral::io::sqlite_data_provider>(sql, 0, 0);
     isSqlProvider = true;
+#else
+      throw std::runtime_error("ERROR: This BlazingSQL version doesn't support SQLite integration");
 #endif
   }
 

diff --git a/engine/src/execution_graph/logic_controllers/BatchProcessing.cpp b/engine/src/execution_graph/logic_controllers/BatchProcessing.cpp
@@ -8,8 +8,13 @@
 #include "io/data_provider/sql/MySQLDataProvider.h"
 #endif
 
-// TODO percy
-//#include "io/data_parser/sql/PostgreSQLParser.h"
+#ifdef POSTGRESQL_SUPPORT
+#include "io/data_provider/sql/PostgreSQLDataProvider.h"
+#endif
+
+#ifdef SQLITE_SUPPORT
+#include "io/data_provider/sql/SQLiteDataProvider.h"
+#endif
 
 #include "parser/expression_utils.hpp"
 #include "taskflow/executor.h"
@@ -128,6 +133,18 @@ TableScan::TableScan(std::size_t kernel_id, const std::string & queryString, std
       ral::io::set_sql_projections<ral::io::mysql_data_provider>(provider.get(), get_projections_wrapper(schema.get_num_columns()));
 #else
       throw std::runtime_error("ERROR: This BlazingSQL version doesn't support MySQL integration");
+#endif
+    } else if (parser->type() == ral::io::DataType::POSTGRESQL)	{
+#ifdef POSTGRESQL_SUPPORT
+      ral::io::set_sql_projections<ral::io::postgresql_data_provider>(provider.get(), get_projections_wrapper(schema.get_num_columns()));
+#else
+      throw std::runtime_error("ERROR: This BlazingSQL version doesn't support PostgreSQL integration");
+#endif
+    } else if (parser->type() == ral::io::DataType::SQLITE)	{
+#ifdef SQLITE_SUPPORT
+      ral::io::set_sql_projections<ral::io::sqlite_data_provider>(provider.get(), get_projections_wrapper(schema.get_num_columns()));
+#else
+      throw std::runtime_error("ERROR: This BlazingSQL version doesn't support SQLite integration");
 #endif
     } else {
         num_batches = provider->get_num_handles();
@@ -237,6 +254,7 @@ BindableTableScan::BindableTableScan(std::size_t kernel_id, const std::string &
 : kernel(kernel_id, queryString, context, kernel_type::BindableTableScanKernel), provider(provider), parser(parser), schema(schema) {
     this->query_graph = query_graph;
     this->filtered = is_filtered_bindable_scan(expression);
+    this->predicate_pushdown_done = false;
 
     if(parser->type() == ral::io::DataType::CUDF || parser->type() == ral::io::DataType::DASK_CUDF){
         num_batches = std::max(provider->get_num_handles(), (size_t)1);
@@ -263,8 +281,21 @@ BindableTableScan::BindableTableScan(std::size_t kernel_id, const std::string &
     } else if (parser->type() == ral::io::DataType::MYSQL)	{
 #ifdef MYSQL_SUPPORT
       ral::io::set_sql_projections<ral::io::mysql_data_provider>(provider.get(), get_projections_wrapper(schema.get_num_columns(), queryString));
+      predicate_pushdown_done = ral::io::set_sql_predicate_pushdown<ral::io::mysql_data_provider>(provider.get(), queryString);
 #else
       throw std::runtime_error("ERROR: This BlazingSQL version doesn't support MySQL integration");
+#endif
+    } else if (parser->type() == ral::io::DataType::POSTGRESQL)	{
+#ifdef POSTGRESQL_SUPPORT
+      ral::io::set_sql_projections<ral::io::postgresql_data_provider>(provider.get(), get_projections_wrapper(schema.get_num_columns(), queryString));
+#else
+      throw std::runtime_error("ERROR: This BlazingSQL version doesn't support PostgreSQL integration");
+#endif
+    } else if (parser->type() == ral::io::DataType::SQLITE)	{
+#ifdef SQLITE_SUPPORT
+      ral::io::set_sql_projections<ral::io::sqlite_data_provider>(provider.get(), get_projections_wrapper(schema.get_num_columns(), queryString));
+#else
+      throw std::runtime_error("ERROR: This BlazingSQL version doesn't support SQLite integration");
 #endif
     } else {
         num_batches = provider->get_num_handles();
@@ -278,7 +309,7 @@ ral::execution::task_result BindableTableScan::do_process(std::vector< std::uniq
     std::unique_ptr<ral::frame::BlazingTable> filtered_input;
 
     try{
-        if(this->filtered) {
+        if(this->filtered && !this->predicate_pushdown_done) {
             filtered_input = ral::processor::process_filter(input->toBlazingTableView(), expression, this->context.get());
             filtered_input->setNames(fix_column_aliases(filtered_input->names(), expression));
             output->addToCache(std::move(filtered_input));

diff --git a/engine/src/execution_graph/logic_controllers/BatchProcessing.h b/engine/src/execution_graph/logic_controllers/BatchProcessing.h
@@ -203,6 +203,7 @@ class BindableTableScan : public kernel {
 	size_t file_index = 0;
 	size_t num_batches;
 	bool filtered;
+  bool predicate_pushdown_done;
 };
 
 /**

diff --git a/engine/src/io/data_parser/ArgsUtil.cpp b/engine/src/io/data_parser/ArgsUtil.cpp
@@ -276,8 +276,8 @@ sql_info getSqlInfo(std::map<std::string, std::string> &args_map) {
   if (args_map.find("password") != args_map.end()) {
     sql.password = args_map.at("password");
   }
-  if (args_map.find("schema") != args_map.end()) {
-    sql.schema = args_map.at("schema");
+  if (args_map.find("database") != args_map.end()) {
+    sql.schema = args_map.at("database");
   }
   if (args_map.find("table") != args_map.end()) {
     sql.table = args_map.at("table");

diff --git a/engine/src/io/data_parser/sql/AbstractSQLParser.cpp b/engine/src/io/data_parser/sql/AbstractSQLParser.cpp
@@ -44,15 +44,35 @@ std::unique_ptr<ral::frame::BlazingTable> abstractsql_parser::parse_batch(
 {
   void *src = nullptr;
 
+  if (type() == DataType::MYSQL) {
 #if defined(MYSQL_SUPPORT)
-  src = handle.sql_handle.mysql_resultset.get();
-#elif defined(POSTGRESQL_SUPPORT)
-  src = handle.sql_handle.postgresql_result.get();
-#elif defined(SQLITE_SUPPORT)
-  src = handle.sql_handle.sqlite_statement.get();
+    src = handle.sql_handle.mysql_resultset.get();
+#else
+    throw std::runtime_error(
+      "Unsupported MySQL parser for this BlazingSQL version");
 #endif
+  }
+
+  if (type() == DataType::POSTGRESQL) {
+#if defined(POSTGRESQL_SUPPORT)
+    src = handle.sql_handle.postgresql_result.get();
+#else
+    throw std::runtime_error(
+      "Unsupported PostgreSQL parser for this BlazingSQL version");
+#endif
+  }
 
-  return this->parse_raw_batch(src, schema, column_indices, row_groups, handle.sql_handle.row_count);
+  if (type() == DataType::SQLITE) {
+#if defined(SQLITE_SUPPORT)
+    src = handle.sql_handle.sqlite_statement.get();
+#else
+    throw std::runtime_error(
+      "Unsupported Sqlite3 parser for this BlazingSQL version");
+#endif
+  }
+
+  return this->parse_raw_batch(
+    src, schema, column_indices, row_groups, handle.sql_handle.row_count);
 }
 
 void abstractsql_parser::parse_schema(ral::io::data_handle handle, ral::io::Schema & schema) {
@@ -277,7 +297,7 @@ std::pair<std::vector<void*>, std::vector<std::vector<cudf::bitmask_type>>> init
       case cudf::type_id::NUM_TYPE_IDS: {} break;
     }
   }
-  
+
   return std::make_pair(host_cols, null_masks);
 }
 

diff --git a/engine/src/io/data_parser/sql/AbstractSQLParser.h b/engine/src/io/data_parser/sql/AbstractSQLParser.h
@@ -1,6 +1,5 @@
 /*
- * Copyright 2021 BlazingDB, Inc.
- *     Copyright 2021 Percy Camilo Triveño Aucahuasi <[email protected]>
+ * Copyright 2021 Percy Camilo Triveño Aucahuasi <[email protected]>
  */
 
 #ifndef _ABSTRACTSQLPARSER_H_