# BigQueryStorage -------------------------------------------------------
auth_fn <- function() {
  tmp <- tempfile(pattern = ".json")
  on.exit(unlink(tmp))
  writeBin(base64enc::base64decode(Sys.getenv("GCP_SERVICE_ACCOUNT")), tmp)
  bigrquery::bq_auth(path = tmp)
}

test_that("BigQuery json and BigQuery return the same results", {
  auth_fn()

  # Compare with bigrquery method
  dt <- bqs_table_download("bigquery-public-data.usa_names.usa_1910_current", bigrquery::bq_test_project(), n_max = 50000, as_tibble = TRUE, quiet = TRUE)
  dt2 <- bigrquery::bq_table_download("bigquery-public-data.usa_names.usa_1910_current", n_max = 50000, quiet = TRUE, api = "json")
  expect_equal(dt, dt2)
})

test_that("Optional BigQuery Storage API parameters work", {
  auth_fn()

  # Check other features
  dt <- bqs_table_download("bigquery-public-data.usa_names.usa_1910_current",
    bigrquery::bq_test_project(),
    selected_fields = c("name", "number", "state"),
    row_restriction = 'state = "WA"',
    quiet = TRUE
  )
  expect_length(names(dt), 3)
  expect_identical(as.character(unique(dt$state)), "WA")

  # BigQuery tables are organized into data blocks. The TABLESAMPLE clause works by randomly
  # selecting a percentage of data blocks from the table and reading all of the rows in the
  # selected blocks. The sampling granularity is limited by the number of data blocks.
  #
  # Typically, BigQuery splits tables or table partitions into blocks if they are larger than
  # about 1 GB. Smaller tables might consist of a single data block. In that case, the TABLESAMPLE
  # clause reads the entire table. If the sampling percentage is greater than zero and the table
  # is not empty, then table sampling always returns some results.
  dt <- bqs_table_download(
    "bigquery-public-data.usa_names.usa_1910_current",
    bigrquery::bq_test_project(),
    selected_fields = "number",
    sample_percentage = 0,
    quiet = TRUE
  )
  expect_true(nrow(dt) == 0)
})

# types -------------------------------------------------------------------

test_that("can read utf-8 strings", {
  auth_fn()

  sql <- "SELECT '\U0001f603' as x"
  tb <- bigrquery::bq_project_query(bigrquery::bq_test_project(), sql, quiet = TRUE)
  df <- bqs_table_download(tb, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE)
  x <- df$x[[1]]

  expect_equal(Encoding(x), "UTF-8")
  expect_equal(x, "\U0001f603")
})

# https://cloud.google.com/bigquery/docs/reference/storage/#arrow_schema_details
# DATETIME does not a have a timezone
test_that("can convert date time types", {
  auth_fn()

  sql <- "SELECT
    datetime,
    CAST (datetime as DATE) as date,
    CAST (datetime as TIME) as time,
    CAST (datetime as TIMESTAMP) as timestamp
    FROM (SELECT DATETIME '2000-01-02 03:04:05.67' as datetime)
  "

  tb <- bigrquery::bq_project_query(bigrquery::bq_test_project(), sql, quiet = TRUE)
  df <- bqs_table_download(tb, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE)
  df2 <- bigrquery::bq_table_download(tb, quiet = TRUE)

  base <- ISOdatetime(2000, 1, 2, 3, 4, 5.67, tz = "UTC")

  expect_equal(df, df2, tolerance = 0.67)
  expect_equal(df$datetime, base)
  expect_equal(df$timestamp, base)
  expect_equal(df$date, as.Date(base))
  expect_equal(df$time, hms::hms(hours = 3, minutes = 4, seconds = 5.67))
})

test_that("correctly parse logical values", {
  auth_fn()

  query <- "SELECT TRUE as x"
  tb <- bigrquery::bq_project_query(bigrquery::bq_test_project(), query)
  df <- bqs_table_download(tb, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE)

  expect_true(df$x)
})

test_that("the return type of integer columns is set by the bigint argument", {
  auth_fn()

  x <- c("-2147483648", "-2147483647", "-1", "0", "1", "2147483647", "2147483648", "18014398509481984", "9007199254740992")
  sql <- paste0("SELECT * FROM UNNEST ([", paste0(x, collapse = ","), "]) AS x")
  qry <- bigrquery::bq_project_query(bigrquery::bq_test_project(), sql)

  expect_warning(
    out_int <- bqs_table_download(qry, bigrquery::bq_test_project(), as_tibble = TRUE, bigint = "integer", quiet = TRUE)$x,
    "NAs produced by integer overflow"
  )
  expect_identical(out_int, suppressWarnings(as.integer(x)))

  x <- c("-2147483648", "-2147483647", "-1", "0", "1", "2147483647", "2147483648")
  sql <- paste0("SELECT * FROM UNNEST ([", paste0(x, collapse = ","), "]) AS x")
  qry <- bigrquery::bq_project_query(bigrquery::bq_test_project(), sql)

  out_int64 <- bqs_table_download(qry, bigrquery::bq_test_project(), as_tibble = TRUE, bigint = "integer64", quiet = TRUE)$x
  expect_identical(out_int64, bit64::as.integer64(x))

  out_dbl <- bqs_table_download(qry, bigrquery::bq_test_project(), as_tibble = TRUE, bigint = "numeric", quiet = TRUE)$x
  expect_identical(out_dbl, as.double(x))

  out_chr <- bqs_table_download(qry, bigrquery::bq_test_project(), as_tibble = TRUE, bigint = "character", quiet = TRUE)$x
  expect_identical(out_chr, x)

})

test_that("float return as numeric", {
  auth_fn()
	x <- c("-2.147483648", "-1.5", "0.5", "1.5")
	sql <- paste0("SELECT * FROM UNNEST ([", paste0(x, collapse = ","), "]) AS x")
	qry <- bigrquery::bq_project_query(bigrquery::bq_test_project(), sql)
	out <- bqs_table_download(qry, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE)$x
	expect_identical(out, as.numeric(x))

})

test_that("n_max returns no more rows than actual originaly in table", {
  auth_fn()
  query <- "SELECT TRUE as x"
  tb <- bigrquery::bq_project_query(bigrquery::bq_test_project(), query)
  df <- bqs_table_download(tb, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE, n_max = 50)
  expect_equal(nrow(df), 1)
})

# Geography is mapped to an utf8 string in input,
# it would have to be converted to a geography by the user
test_that("can convert geography type", {
  auth_fn()

  skip_if_not_installed("wk")
  sql <- "SELECT ST_GEOGFROMTEXT('POINT (30 10)') as geography"
  tb <- bigrquery::bq_project_query(bigrquery::bq_test_project(), sql, quiet = TRUE)
  df <- bqs_table_download(tb, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE)

  expect_identical(df$geography, wk::wkt("POINT(30 10)"))
})

test_that("can convert bytes type", {
  auth_fn()

  sql <- "SELECT ST_ASBINARY(ST_GEOGFROMTEXT('POINT (30 10)')) as bytes"
  tb <- bigrquery::bq_project_query(bigrquery::bq_test_project(), sql, quiet = TRUE)
  df <- bqs_table_download(tb, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE)

  expect_identical(
    df$bytes,
    blob::as_blob(as.raw(c(
      0x01, 0x01, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff,
      0xff, 0xff, 0x3d, 0x40, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x24,
      0x40
    )))
  )
})

# https://github.com/meztez/bigrquerystorage/issues/81
# RANGE columns previously errored in parse_postprocess()/col_mapply() because
# the RANGE struct has 2 sub-fields (start, end) but the underlying record
# handling assumed the named vector length always matched y[["fields"]].
test_that("can download RANGE type", {
  auth_fn()

  sql <- "SELECT RANGE(DATE '2024-01-01', DATE '2025-01-01') AS year_range"
  tb <- bigrquery::bq_project_query(bigrquery::bq_test_project(), sql, quiet = TRUE)

  expect_no_error(
    df <- bqs_table_download(tb, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE)
  )

  # RANGE is returned as a start/end struct of Dates (the arrow api has no
  # native RANGE type, unlike the json api which returns a "[start, end)"
  # string); check the decoded boundaries match the json api's values.
  expect_equal(nrow(df), 1)
  expect_true("year_range" %in% names(df))
  expect_true(inherits(df$year_range, "data.frame"))
  expect_setequal(names(df$year_range), c("start", "end"))
  expect_equal(df$year_range$start, as.Date("2024-01-01"))
  expect_equal(df$year_range$end, as.Date("2025-01-01"))
})

test_that("nested list type", {
	auth_fn()
	sql <- "
	  SELECT STRUCT(1 AS a, 'abc' AS b) as s,
           [1, 2, 3] as a,
           [STRUCT(1 as a, 'a' as b), STRUCT(2, 'b'), STRUCT(3, 'c')] as aos,
           STRUCT([1, 2, 3] as a, ['a', 'b'] as b) as soa
  "

	tb <- bigrquery::bq_project_query(bigrquery::bq_test_project(), sql, quiet = TRUE)
	df <- bqs_table_download(tb, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE)

	expect_equal(df[["s"]], tibble::tibble(a = 1, b = "abc"))
	expect_equal(df[["a"]], list(1:3))
	expect_equal(df[["aos"]], list(tibble::tibble(a = 1:3, b = c("a", "b", "c"))))
	expect_equal(df[["soa"]], tibble::tibble(a = list(1:3), b = list(c("a", "b"))))

})

test_that("post process parse works", {
	auth_fn()
	sql <- "
	SELECT
    '\U0001f603' as unicode,
    datetime,
    TRUE as logicaltrue,
    FALSE as logicalfalse,
    CAST ('Hi' as BYTES) as bytes,
    CAST (datetime as DATE) as date,
    CAST (datetime as TIME) as time,
    CAST (datetime as TIMESTAMP) as timestamp,
    ST_GEOGFROMTEXT('POINT (30 10)') as geography,
    STRUCT(1 AS a, 'abc' AS b) as s,
    [1, 2, 3] as a,
    [STRUCT(1 as a, ST_GEOGFROMTEXT('POINT (5 5)') as b), STRUCT(2, ST_GEOGFROMTEXT('POINT (10 10)')), STRUCT(3, ST_GEOGFROMTEXT('POINT (15 15)'))] as aos,
    STRUCT([1, 2, 3] as a, ['a', 'b'] as b) as soa,
    STRUCT([CAST ('Hi' as BYTES), CAST ('Bob' as BYTES)] as a, ['a', 'b'] as b) as bb,
    STRUCT([ST_GEOGFROMTEXT('POINT (30 10)'), ST_GEOGFROMTEXT('POINT (15 15)')] as geo) as gg
  FROM (SELECT DATETIME '2000-01-02 03:04:05.67' as datetime)"
	tb <- bigrquery::bq_project_query(bigrquery::bq_test_project(), sql, quiet = TRUE)
	df <- bqs_table_download(tb, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE)
	df2 <- bqs_table_download(tb, bigrquery::bq_test_project(), as_tibble = TRUE, quiet = TRUE, selected_fields = c("bb.a", "uniCODE", "aos"))
  expect_equal(attr(df$datetime,"tzone"), "UTC")
  expect_true(inherits(df$bytes, "blob"))
  expect_true(inherits(df$geography, "wk_wkt"))
  expect_true(inherits(df$aos[[1]]$b, "wk_wkt"))
  expect_true(inherits(df$bb$a[[1]], "blob"))
  expect_true(inherits(df$gg$geo[[1]], "wk_wkt"))
  expect_equal(length(df2), 3)
})