Datasource with random data.

database.go

@@ -0,0 +1,138 @@
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "sort"
+ "time"
+
+ "github.com/cube2222/octosql/octosql"
+ _ "github.com/jackc/pgx/stdlib"
+
+ "github.com/cube2222/octosql/physical"
+ "github.com/cube2222/octosql/plugins/plugin"
+)
+
+var tableInternalNames = map[string]string{
+ "addresses": "address/random_address",
+ "companies": "company/random_company",
+ "users": "users/random_user",
+}
+
+func Creator(ctx context.Context, configUntyped plugin.ConfigDecoder) (physical.Database, error) {
+ return &Database{}, nil
+}
+
+type Database struct {
+}
+
+func (d *Database) ListTables(ctx context.Context) ([]string, error) {
+ return []string{
+ "addresses",
+ "companies",
+ "users",
+ }, nil
+}
+
+func (d *Database) GetTable(ctx context.Context, name string) (physical.DatasourceImplementation, physical.Schema, error) {
+ internalName, ok := tableInternalNames[name]
+ if !ok {
+ return nil, physical.Schema{}, fmt.Errorf("table %s not found", name)
+ }
+
+ req, err := http.NewRequest("GET", fmt.Sprintf("https://random-data-api.com/api/%s", internalName), nil)
+ if err != nil {
+ return nil, physical.Schema{}, fmt.Errorf("couldn't create request to get sample data: %w", err)
+ }
+ req = req.WithContext(ctx)
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ return nil, physical.Schema{}, fmt.Errorf("couldn't get sample data: %w", err)
+ }
+ defer res.Body.Close()
+
+ var data map[string]interface{}
+ if err := json.NewDecoder(res.Body).Decode(&data); err != nil {
+ return nil, physical.Schema{}, fmt.Errorf("couldn't decode sample data: %w", err)
+ }
+
+ fieldNames := make([]string, 0, len(data))
+ for k := range data {
+ fieldNames = append(fieldNames, k)
+ }
+ sort.Strings(fieldNames)
+
+ fields := make([]physical.SchemaField, len(fieldNames))
+ for i := range fieldNames {
+ fields[i] = physical.SchemaField{
+ Name: fieldNames[i],
+ Type: getOctoSQLType(data[fieldNames[i]]),
+ }
+ }
+
+ return &datasourceImpl{
+ internalName: internalName,
+ },
+ physical.Schema{
+ Fields: fields,
+ TimeField: -1,
+ },
+ nil
+}
+
+func getOctoSQLType(value interface{}) octosql.Type {
+ switch value := value.(type) {
+ case int:
+ return octosql.Int
+ case bool:
+ return octosql.Boolean
+ case float64:
+ return octosql.Float
+ case string:
+ return octosql.String
+ case time.Time:
+ return octosql.Time
+ case map[string]interface{}:
+ fieldNames := make([]string, 0, len(value))
+ for k := range value {
+ fieldNames = append(fieldNames, k)
+ }
+ sort.Strings(fieldNames)
+ fields := make([]octosql.StructField, len(value))
+ for i := range fieldNames {
+ fields[i] = octosql.StructField{
+ Name: fieldNames[i],
+ Type: getOctoSQLType(value[fieldNames[i]]),
+ }
+ }
+ return octosql.Type{
+ TypeID: octosql.TypeIDStruct,
+ Struct: struct{ Fields []octosql.StructField }{Fields: fields},
+ }
+ case []interface{}:
+ var elementType *octosql.Type
+ for i := range value {
+ if elementType != nil {
+ t := octosql.TypeSum(*elementType, getOctoSQLType(value[i]))
+ elementType = &t
+ } else {
+ t := getOctoSQLType(value[i])
+ elementType = &t
+ }
+ }
+ return octosql.Type{
+ TypeID: octosql.TypeIDList,
+ List: struct {
+ Element *octosql.Type
+ }{
+ Element: elementType,
+ },
+ }
+ case nil:
+ return octosql.Null
+ }
+
+ panic(fmt.Sprintf("unexhaustive json input value match: %T %+v", value, value))
+}

execution.go

@@ -0,0 +1,68 @@
+package main
+
+import (
+ "fmt"
+ "time"
+
+ . "github.com/cube2222/octosql/execution"
+ "github.com/cube2222/octosql/octosql"
+ "github.com/cube2222/octosql/physical"
+)
+
+type datasourceExecuting struct {
+ data []map[string]interface{}
+ fields []physical.SchemaField
+}
+
+func (d *datasourceExecuting) Run(ctx ExecutionContext, produce ProduceFn, metaSend MetaSendFn) error {
+ for i := range d.data {
+ values := make([]octosql.Value, len(d.fields))
+ for j, field := range d.fields {
+ values[j] = getOctoSQLValue(field.Type, d.data[i][field.Name])
+ }
+
+ if err := produce(
+ ProduceFromExecutionContext(ctx),
+ NewRecord(values, false, time.Time{}),
+ ); err != nil {
+ return fmt.Errorf("couldn't produce record: %w", err)
+ }
+ }
+
+ return nil
+}
+
+func getOctoSQLValue(t octosql.Type, value interface{}) octosql.Value {
+ switch value := value.(type) {
+ case int:
+ return octosql.NewInt(value)
+ case bool:
+ return octosql.NewBoolean(value)
+ case float64:
+ return octosql.NewFloat(value)
+ case string:
+ return octosql.NewString(value)
+ case time.Time:
+ return octosql.NewTime(value)
+ case map[string]interface{}:
+ values := make([]octosql.Value, len(t.Struct.Fields))
+ for i, field := range t.Struct.Fields {
+ values[i] = getOctoSQLValue(field.Type, value[field.Name])
+ }
+ return octosql.NewStruct(values)
+ case []interface{}:
+ if t.List.Element == nil && len(value) > 0 {
+ panic("list was expected to be empty (typeless) but isn't")
+ }
+
+ elements := make([]octosql.Value, len(value))
+ for i := range elements {
+ elements[i] = getOctoSQLValue(*t.List.Element, value[i])
+ }
+ return octosql.NewList(elements)
+ case nil:
+ return octosql.NewNull()
+ }
+
+ panic(fmt.Sprintf("unexhaustive json input value match: %T %+v", value, value))
+}

go.mod

@@ -0,0 +1,27 @@
+module github.com/cube2222/octosql-plugin-random_data
+
+require (
+ github.com/cube2222/octosql v0.3.1-0.20211211163254-02606a8a60a6
+ github.com/davecgh/go-spew v1.1.1
+ github.com/jackc/pgx v3.6.2+incompatible
+)
+
+require (
+ github.com/awalterschulze/gographviz v2.0.3+incompatible // indirect
+ github.com/cespare/xxhash v1.1.0 // indirect
+ github.com/dgraph-io/ristretto v0.0.3 // indirect
+ github.com/golang/protobuf v1.5.2 // indirect
+ github.com/google/btree v1.0.0 // indirect
+ github.com/oklog/ulid/v2 v2.0.2 // indirect
+ github.com/pkg/errors v0.9.1 // indirect
+ golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad // indirect
+ golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2 // indirect
+ golang.org/x/sys v0.0.0-20211113001501-0c823b97ae02 // indirect
+ golang.org/x/text v0.3.7 // indirect
+ google.golang.org/genproto v0.0.0-20211112145013-271947fe86fd // indirect
+ google.golang.org/grpc v1.42.0 // indirect
+ google.golang.org/protobuf v1.27.1 // indirect
+ gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
+)
+
+go 1.17