all: implement Spanner connector

[odeke-em]

Allows us to build SQL from filters and required columns.

Updates #58

[halio-g]

/gcbrun

[halio-g]

/gcbrun

[halio-g]

/gcbrun

[halio-g]

/gcbrun

[halio-g]

Can we add an integration test (spark session) one to test it ? Cause last time when I tested the range in the real dataproc job.

"2022-12-31" results in 19357 but really it should be 1672444800000000

Sorry, I misunderstood your question, it should be 19357 since I think spark receive the # of days from 1970-01-01 instead of microseconds.

[odeke-em]

Can we add an integration test (spark session) one to test it ? Cause last time when I tested the range in the real dataproc job.

"2022-12-31" results in 19357 but really it should be 1672444800000000

Sorry, I misunderstood your question, it should be 19357 since I think spark receive the # of days from 1970-01-01 instead of microseconds.

No worries, over the weekend I reverted to return days since the epoch.

[odeke-em]

@davidrabinowitz @halio-g at current commit, these are the results from running this Python program

from pyspark.sql import SparkSession

def main():
    table = "exchange_rates"
    spark = SparkSession.builder.appName("Query Spanner").getOrCreate()
    df = spark.read.format('cloud-spanner') \
                .option("projectId", PROJECT_ID) \
                .option("instanceId", SPANNER_INSTANCE) \
                .option("databaseId", SPANNER_DATABASE) \
                .option("enableDataBoost", "true") \
                .option("table", table) \
                .load(table)
    df.printSchema()
    df.select("created_at", "value", "base_cur").filter((df["value"] > 3720) & (df["base_cur"] == "USD")).show()

which produces

root
 |-- id: string (nullable = false)
 |-- base_cur: string (nullable = false)
 |-- end_cur: string (nullable = false)
 |-- value: double (nullable = false)
 |-- data_src: string (nullable = false)
 |-- created_at: timestamp (nullable = false)
 |-- published_at: timestamp (nullable = true)

+--------------------+-----------+--------+
|          created_at|      value|base_cur|
+--------------------+-----------+--------+
|2023-08-30 20:56:...|3724.365901|     USD|
|2023-09-09 04:40:...|3738.384066|     USD|
|2023-09-07 05:44:...|3735.182945|     USD|
|2023-08-25 17:00:...|3730.606064|     USD|
|2023-09-15 10:08:...|3724.849835|     USD|
|2023-08-21 18:24:...|3727.353643|     USD|
|2023-08-20 07:52:...|3724.922281|     USD|
|2023-08-23 23:00:...|3738.342818|     USD|
|2023-08-14 10:00:...|3749.635375|     USD|
|2023-09-01 05:48:...|3730.472718|     USD|
|2023-09-15 21:40:...|3730.465981|     USD|
|2023-09-11 16:40:...|3723.400239|     USD|
|2023-09-13 13:16:...|3723.537745|     USD|
|2023-08-19 14:00:...|3724.922281|     USD|
|2023-09-06 02:56:...|3735.927727|     USD|

[halio-g]

Please fix the presubmit issue. https://github.com/GoogleCloudDataproc/spark-spanner-connector/pull/59/checks?check_run_id=16914932023

[odeke-em]

@halio-g the presubmit check now passes all gucci, please take a look again!

Addressed almost all the comments, please re-review.

[halio-g]

This can be done in a follow up pr.

Please create table by types, for example:
CREATE TABLE IntTable(
not_null_index INT64 NOT NULL,
int_column INT64,
) PRIMARY KEY(not_null_index);

CREATE TABLE StringTable(
not_null_index STRING<10> NOT NULL,
max_string_col STRING,
max_string_col STRING<100>,
) PRIMARY KEY(not_null_index);

CREATE TABLE TimestampTable(
not_null_index TIMESTAMP NOT NULL,
timestamp_col TIMESTAMP,
) PRIMARY KEY(not_null_index);

Or creating a single table with all the types, for example:
CREATE TABLE IntegrationTestTable(
not_null_index INT64 NOT NULL,
max_string_col STRING,
100_string_col STRING<100>,
int_column INT64,
timestamp_col TIMESTAMP,
) PRIMARY KEY(not_null_index);

CREATE TABLE StringIndexTable(
string_index STRING<20> NOT NULL,
) PRIMARY KEY(string_index);