GoogleCloudPlatform / GoogleCloudPlatform/cloud-spanner-emulator

TimestampBounds using ReadTimestamps do not behave as expected

Open
#189 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
C++
Stars
334
Forks
77
Avg merge
8m
Merged PRs (30d)
2

Description

TimestampBounds do not work as expected in the emulator (this behavior does work as expected using Cloud Spanner). Using a ReadTimestamp of a time shortly after the committed time does not return the previously inserted rows. Increasing a delay before reading does return the values that were inserted.

code sample to reproduce:
```Go
package main

import (
"cloud.google.com/go/spanner"
database "cloud.google.com/go/spanner/admin/database/apiv1"
"cloud.google.com/go/spanner/admin/database/apiv1/databasepb"
instance "cloud.google.com/go/spanner/admin/instance/apiv1"
"cloud.google.com/go/spanner/admin/instance/apiv1/instancepb"
"context"
"fmt"
_ "github.com/googleapis/go-sql-spanner"
"google.golang.org/api/iterator"
"google.golang.org/grpc/codes"
"os"
"os/signal"
"time"
)

func main() {
err := run()
if err != nil {
fmt.Fprint(os.Stderr, err)
os.Exit(1)
}
}

func run() error {
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
defer cancel()

const (
ProjectID, InstanceID, DatabaseID, DatabaseNameEscaped = "your-project-id", "test-instance", "testdb", "`testdb`"

Database = "projects/" + ProjectID + "/instances/" + InstanceID + "/databases/" + DatabaseID
)

{ // create a new instance
admin, err := instance.NewInstanceAdminClient(ctx)
if err != nil {
return fmt.Errorf("failed to create instance admin: %w", err)
}
op, err := admin.CreateInstance(ctx, &instancepb.CreateInstanceRequest{
Parent: "projects/" + ProjectID,
InstanceId: InstanceID,
})
if err != nil {
if spanner.ErrCode(err) == codes.AlreadyExists {
goto instanceReady
}
return fmt.Errorf("create instance failed: %w", err)
}
if _, err := op.Wait(ctx); err != nil {
return fmt.Errorf("failed to wait instance creation: %w", err)
}

instanceReady:
if err := admin.Close(); err != nil {
return fmt.Errorf("failed to close instance admin: %w", err)
}
}

{ // create a database
admin, err := database.NewDatabaseAdminClient(ctx)
if err != nil {
return fmt.Errorf("failed to create database admin: %w", err)
}

ddl, err := admin.CreateDatabase(ctx, &databasepb.CreateDatabaseRequest{
Parent: "projects/" + ProjectID + "/instances/" + InstanceID,
CreateStatement: "CREATE DATABASE " + DatabaseNameEscaped,
ExtraStatements: []string{},
DatabaseDialect: databasepb.DatabaseDialect_GOOGLE_STANDARD_SQL,
})
if err != nil {
if spanner.ErrCode(err) == codes.AlreadyExists {
goto databaseReady
}
return fmt.Errorf("failed to create database: %w", err)
}

if _, err := ddl.Wait(ctx); err != nil {
return fmt.Errorf("failed to wait database creation: %w", err)
}

databaseReady:
updateReq := &databasepb.UpdateDatabaseDdlRequest{
Database: Database,
Statements: []string{
`CREATE TABLE IF NOT EXISTS testtime (
id INT64 NOT NULL,
value TIMESTAMP NOT NULL
) PRIMARY KEY(id)`,
},
}

updateDDL, err := admin.UpdateDatabaseDdl(ctx, updateReq)
if err != nil {
return fmt.Errorf("failed to update database ddl: %w", err)
}

if err := updateDDL.Wait(ctx); err != nil {
return fmt.Errorf("failed to wait for database ddl: %w", err)
}

if err := admin.Close(); err != nil {
return fmt.Errorf("failed to close database admin: %w", err)
}

db, err := openDatabaseConn(ctx, Database)
if err != nil {
return fmt.Errorf("failed to open database connection: %w", err)
}
defer db.Close()

_, err = db.ReadWriteTransaction(ctx, func(ctx context.Context, tx *spanner.ReadWriteTransaction) error {
_, err = tx.BatchUpdate(ctx, []spanner.Statement{
{SQL: "INSERT INTO testtime (id, value) VALUES (1, @value)", Params: map[string]interface{}{"value": time.Now()}},
{SQL: "INSERT INTO testtime (id, value) VALUES (2, @value)", Params: map[string]interface{}{"value": time.Now().Add(time.Hour)}},
{SQL: "INSERT INTO testtime (id, value) VALUES (3, @value)", Params: map[string]interface{}{"value": time.Now().Add(2 * time.Hour)}},
})
return err
})
if err != nil {
return fmt.Errorf("failed to insert values into testtime: %w", err)
}

fmt.Println("committed data")

// wait a short bit of time
time.Sleep(1 * time.Second)

fmt.Println("data after short delay but after expected commit time")

// data will not print in the emulator using a short time after the expected committed time
if err := readAtTimestampAndPrintData(ctx, db, time.Now().Add(-500*time.Millisecond)); err != nil {
return fmt.Errorf("failed to read data at timestamp: %w", err)
}

// wait longer to ensure the data is truly committed
time.Sleep(3 * time.Second)

fmt.Println("data after longer delay after expected commit time")

// data will print in the emulator as the delay is long enough
if err := readAtTimestampAndPrintData(ctx, db, time.Now().Add(-500*time.Millisecond)); err != nil {
return fmt.Errorf("failed to read data at timestamp: %w", err)
}

_, err = db.ReadWriteTransaction(ctx, func(ctx context.Context, tx *spanner.ReadWriteTransaction) error {
_, err = tx.Update(ctx, spanner.Statement{
SQL: "DELETE FROM testtime WHERE 1=1",
})
return err
})
if err != nil {
return fmt.Errorf("failed to delete values from testtime: %w", err)
}

return nil
}
}

func openDatabaseConn(ctx context.Context, databaseConn string) (*spanner.Client, error) {
client, err := spanner.NewClient(ctx, databaseConn)
if err != nil {
return nil, fmt.Errorf("failed to create spanner client: %w", err)
}
return client, nil
}

func readAtTimestampAndPrintData(ctx context.Context, db *spanner.Client, timestamp time.Time) error {
it := db.Single().WithTimestampBound(spanner.ReadTimestamp(timestamp)).Read(ctx, "testtime", spanner.AllKeys(), []string{"id", "value"})
defer it.Stop()

for {
row, err := it.Next()
if err != nil && err != iterator.Done {
return fmt.Errorf("failed to get next object: %w", err)
}
if err == iterator.Done {
break
}
fmt.Println(row.String())
}
return nil
}
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.