Support batch query with parameter

[jingchen2222]

Motivation

In database management systems (DBMS), a prepared statement or parameterized statement is a feature used to execute the same or similar database statements repeatedly with high efficiency. Typically used with SQL statements such as queries or updates, the prepared statement takes the form of a template into which certain constant values are substituted during each execution.[1]

As compared to executing statements directly, prepared statements offer two main advantages:

• The overhead of compiling the statement is incurred only once, although the statement is executed multiple times.
• Prepared statements are resilient against SQL injection because values which are transmitted later using a different protocol are not compiled like the statement template.

We are trying to support query with given parameters under batch-mode

Example

This example uses Java and JDBC for MYSQL database.

import com.mysql.jdbc.jdbc2.optional.MysqlDataSource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

public class Main {

  public static void main(String[] args) throws SQLException {
    MysqlDataSource ds = new MysqlDataSource();
    ds.setDatabaseName("mysql");
    ds.setUser("root");

    try (Connection conn = ds.getConnection()) {
      try (Statement stmt = conn.createStatement()) {
        stmt.executeUpdate("CREATE TABLE IF NOT EXISTS products (name VARCHAR(40), price INT)");
      }

      try (PreparedStatement stmt = conn.prepareStatement("INSERT INTO products VALUES (?, ?)")) {
        stmt.setString(1, "bike");
        stmt.setInt(2, 10900);
        stmt.executeUpdate();
        stmt.setString(1, "shoes");
        stmt.setInt(2, 7400);
        stmt.executeUpdate();
        stmt.setString(1, "phone");
        stmt.setInt(2, 29500);
        stmt.executeUpdate();
      }

      try (PreparedStatement stmt = conn.prepareStatement("SELECT * FROM products WHERE name = ?")) {
        stmt.setString(1, "shoes");
        ResultSet rs = stmt.executeQuery();
        rs.next();
        System.out.println(rs.getInt(2));
      }
    }
  }
}

HybridSE currently support using PrepareStatement to INSERT rows into the table. In this article, we are trying to support query SQL with PrepareStatement.

Design & Implementation

SQL Syntax

Parameterized query statement has been supported in ZETASQL parser.

SELECT card_id, sum(col1) as sum_col1, max(col2) as max_col2, min(col3) as min_col3 FROM t1
WHERE card_id = ? and std_ts >= ? and std_ts <= ? GROUP by card_id having card_id = ?; 

Parameter Representation.

Resolved By https://github.com/4paradigm/HybridSE/pull/182

Data structure

We are going to use ParameterExpr Node to represent a placeholder instead of using a ConstNode .

ParamterExpr node maintains a field position_ to identify the parameter's position (index).

Convert Parameterized QueryStatement into SQL trees and Logical Plan.

The implementation is quite straight forward. Convert zetasql::ASTParameterExpr to ParamterExpr.

Check ISSUE #181 for more details.

Implement Plan

Resolved by https://github.com/4paradigm/HybridSE/pull/186

Schema Infer for Query with parameters types.

• Implement InferAttr() method for ParamterExpr: extract parameter types from parameter_schema_ which has been store in Plan context.
• Update SQL compile related interface.
• Passing parameter schema to SqlContext.

CodeGen https://github.com/4paradigm/HybridSE/pull/187

Modify RowFnLetIRBuilder

• Add @parameter argurement to function let.
• Store and Get @paramter value

Build LLVM IR for ParameterExpr

• Extract ParameterExpr value from @parameter row.
• Add test funclet with @parameter

Engine and Runner #262

Runner Implementation

• BatchRunSession
  We have to support query statements with parameters which means we should design and implement interface to run with given parameter row.

/// \brief BatchRunSession is a kind of RunSession designed for batch mode query.
class BatchRunSession : public RunSession {
 public:
    explicit BatchRunSession(bool mini_batch = false)
        : RunSession(kBatchMode), mini_batch_(mini_batch) {}
    ~BatchRunSession() {}
    /// \brief Query sql in batch mode.
    /// Query results will be returned as std::vector<Row> in output
    int32_t Run(std::vector<Row>& output,  // NOLINT
                uint64_t limit = 0);
    /// \brief Query sql in batch mode.
    /// Query results will be returned as std::vector<Row> in output
    int32_t Run(const Row& input, 
                std::vector<Row>& output,  // NOLINT
                uint64_t limit = 0);

 private:
    const bool mini_batch_;
};

SDK

SQL router #248

Demo

// ... ignore
    hybridse::codec::Schema parameter_schema;
    {
        auto column = parameter_schema.Add();
        column->set_type(::hybridse::type::kVarchar);
    }
    {
        auto column = parameter_schema.Add();
        column->set_type(::hybridse::type::kInt64);
    }
    const std::shared_ptr<::hybridse::sdk::Schema> schema_impl = std::make_shared<hybridse::sdk::SchemaImpl>(schema);
    std::string where_exist = "select * from trans where merch_id = ? and txn_time < ?;";
    // parameterized query
    auto parameter_row = std::make_shared<SQLRequestRow>(schema_impl, std::set<std::string>());
    {
        ASSERT_EQ(2, parameter_row->GetSchema()->GetColumnCnt());
        ASSERT_TRUE(parameter_row->Init(4));
        ASSERT_TRUE(parameter_row->AppendString("mc_0"));
        ASSERT_TRUE(parameter_row->AppendInt64(1594800959830));
        ASSERT_TRUE(parameter_row->Build());

        auto rs = router->ExecuteSQL(db, where_exist, parameter_row, &status);

        if (!rs) {
            FAIL() << "fail to execute sql";
        }
        ASSERT_EQ(rs->Size(), 3);
    }

JDBC #167

// parameterized query
  String parameterizedQuerySql = "select col1, col2 from tsql1010 where col2 = ? and col1 < ?;";
  PreparedStatement query_statement = router.getPreparedStatement(dbname, parameterizedQuerySql);
  // col2 = "hi" and col1 < 1003
  {
      query_statement.setString(1, "hi");
      query_statement.setLong(2, 1003);
      com._4paradigm.openmldb.jdbc.SQLResultSet rs4 = (com._4paradigm.openmldb.jdbc.SQLResultSet) query_statement.executeQuery();
      Assert.assertEquals(2, rs4.GetInternalSchema().GetColumnCnt());
      Assert.assertEquals("kTypeInt64", rs4.GetInternalSchema().GetColumnType(0).toString());
      Assert.assertEquals("kTypeString", rs4.GetInternalSchema().GetColumnType(1).toString());
      Assert.assertTrue(rs4.next());
      Assert.assertEquals(1002, rs4.getLong(1));
      Assert.assertEquals("hi", rs4.getString(2));
      Assert.assertFalse(rs4.next());
      rs4.close();
  }

Performance

Reference

https://en.wikipedia.org/wiki/Prepared_statement