PrestoExample.R

PrestoExample.R

library(RPresto)
library(httr)
library(DBI)

# Utils ####
.request.headers <- function(conn, transaction_id = 'NONE') {
  if(transaction_id == 'NONE'){
    return(httr::add_headers(
      "X-Presto-User"= conn@user,
      "X-Presto-Catalog"= conn@catalog,
      "X-Presto-Schema"= conn@schema,
      "X-Presto-Source"= conn@source,
      "X-Presto-Time-Zone" = conn@session.timezone,
      "X-Presto-Transaction-Id" = 'NONE',
      "User-Agent"= getPackageName(),
      "X-Presto-Session"=conn@session$parameterString()
    ))
  } else {
  return(httr::add_headers(
    "X-Presto-User"= conn@user,
    "X-Presto-Catalog"= conn@catalog,
    "X-Presto-Schema"= conn@schema,
    "X-Presto-Source"= conn@source,
    "X-Presto-Time-Zone" = conn@session.timezone,
    "X-Presto-Transaction-Id" = transaction_id,
    "User-Agent"= getPackageName(),
    "X-Presto-Session"=conn@session$parameterString()
  ))
  }
}

conn = DBI::dbConnect(
  RPresto::Presto(),
  host = host,
  port = 8081,
  user = "Testing",
  schema = 'default',
  catalog = 'hive'
)

initial_statement = 'START TRANSACTION'
next_statement = "CREATE EXAMPLE_TABLE AS (SELECT * FROM {EXAMPLE_TABLE} WHERE date = '2018-10-01' LIMIT 1)")
final_statment = "SELECT * FROM EXAMPLE_TABLE")
clear_statement = "ROLLBACK")

first_url <- paste0(conn@host, ':', conn@port, '/v1/statement')
headers <- .request.headers(conn)
post.response <- httr::POST(first_url, body=enc2utf8(initial_statement), headers)
info_url = httr::content(post.response)$infoUri; 
next_url = paste0(conn@host, ':', conn@port, '/v1/query/', gsub(info_url, pattern='.*\\?', replacement = ''))
resp =httr::GET(next_url);
resp.content = httr::content(resp)

new_headers = .request.headers(conn, as.character(resp.content$startedTransactionId))
post.response <- httr::POST(first_url, body=enc2utf8(next_statement), headers)
post.content = httr::content(post.response)