Spanner to v3 (#404)

* e2e env

* schema

* return subtype with schema

* aggregate, upsert

* support nested queries

* partially error handling

* add to ci

* add default value for upsert

.github/workflows/main.yml

@@ -41,6 +41,7 @@ jobs:
                    "postgres", "postgres13", "postgres12", "postgres11", "postgres10", "postgres9",
                    "mysql", "mysql5",
                    "mssql", "mssql17",
+                   spanner
                    ]
 
     env:

apps/velo-external-db/src/storage/factory.ts

@@ -7,10 +7,10 @@ export const engineConnectorFor = async(_type: string, config: any): Promise<Dat
             const { postgresFactory } = require('@wix-velo/external-db-postgres')
             return await postgresFactory(config)
         }
-        // case 'spanner': {
-        //     const { spannerFactory } = require('@wix-velo/external-db-spanner')
-        //     return await spannerFactory(config)
-        // }
+        case 'spanner': {
+            const { spannerFactory } = require('@wix-velo/external-db-spanner')
+            return await spannerFactory(config)
+        }
         // case 'firestore': {
         //     const { firestoreFactory } = require('@wix-velo/external-db-firestore')
         //     return await firestoreFactory(config)

apps/velo-external-db/test/env/env.db.setup.js

@@ -5,7 +5,7 @@ registerTsProject('.', 'tsconfig.base.json')
 
 const { testResources: postgres } = require ('@wix-velo/external-db-postgres')
 const { testResources: mysql } = require ('@wix-velo/external-db-mysql')
-// const { testResources: spanner } = require ('@wix-velo/external-db-spanner')
+const { testResources: spanner } = require ('@wix-velo/external-db-spanner')
 // const { testResources: firestore } = require ('@wix-velo/external-db-firestore')
 const { testResources: mssql } = require ('@wix-velo/external-db-mssql')
 // const { testResources: mongo } = require ('@wix-velo/external-db-mongo')
@@ -23,9 +23,9 @@ const initEnv = async(testEngine) => {
             await mysql.initEnv()
             break
 
-        // case 'spanner':
-        //     await spanner.initEnv()
-        //     break
+        case 'spanner':
+            await spanner.initEnv()
+            break
 
         case 'postgres':
             await postgres.initEnv()
@@ -66,9 +66,9 @@ const cleanup = async(testEngine) => {
             await mysql.cleanup()
             break
 
-        // case 'spanner':
-        //     await spanner.cleanup()
-        //     break
+        case 'spanner':
+            await spanner.cleanup()
+            break
 
         case 'postgres':
             await postgres.cleanup()

apps/velo-external-db/test/env/env.db.teardown.js

@@ -1,6 +1,6 @@
 const { testResources: postgres } = require ('@wix-velo/external-db-postgres')
 const { testResources: mysql } = require ('@wix-velo/external-db-mysql')
-// const { testResources: spanner } = require ('@wix-velo/external-db-spanner')
+const { testResources: spanner } = require ('@wix-velo/external-db-spanner')
 // const { testResources: firestore } = require ('@wix-velo/external-db-firestore')
 const { testResources: mssql } = require ('@wix-velo/external-db-mssql')
 // const { testResources: mongo } = require ('@wix-velo/external-db-mongo')
@@ -17,9 +17,9 @@ const shutdownEnv = async(testEngine) => {
             await mysql.shutdownEnv()
             break
 
-        // case 'spanner':
-        //     await spanner.shutdownEnv()
-        //     break
+        case 'spanner':
+            await spanner.shutdownEnv()
+            break
 
         case 'postgres':
             await postgres.shutdownEnv()

apps/velo-external-db/test/resources/e2e_resources.ts

@@ -30,7 +30,7 @@ const createAppWithWixDataBaseUrl = createApp.bind(null, wixDataBaseUrl())
 const testSuits = {
     mysql: new E2EResources(mysql, createAppWithWixDataBaseUrl),
     postgres: new E2EResources(postgres, createAppWithWixDataBaseUrl),
-    spanner: new E2EResources(spanner, createApp),
+    spanner: new E2EResources(spanner, createAppWithWixDataBaseUrl),
     firestore: new E2EResources(firestore, createApp),
     mssql: new E2EResources(mssql, createAppWithWixDataBaseUrl),
     mongo: new E2EResources(mongo, createApp),

apps/velo-external-db/test/resources/provider_resources.ts

@@ -71,7 +71,7 @@ const googleSheetTestEnvInit = async() => await dbInit(googleSheet)
 const testSuits = {
     mysql: suiteDef('MySql', mysqlTestEnvInit, mysql.testResources),
     postgres: suiteDef('Postgres', postgresTestEnvInit, postgres.testResources),
-    spanner: suiteDef('Spanner', spannerTestEnvInit, spanner.testResources.supportedOperations),
+    spanner: suiteDef('Spanner', spannerTestEnvInit, spanner.testResources),
     firestore: suiteDef('Firestore', firestoreTestEnvInit, firestore.testResources.supportedOperations),
     mssql: suiteDef('Sql Server', mssqlTestEnvInit, mssql.testResources),
     mongo: suiteDef('Mongo', mongoTestEnvInit, mongo.testResources.supportedOperations),

libs/external-db-spanner/src/spanner_capabilities.ts

@@ -0,0 +1,21 @@
+import { AdapterOperators } from '@wix-velo/velo-external-db-commons'
+import { CollectionOperation, DataOperation, FieldType } from '@wix-velo/velo-external-db-types'
+
+const { query, count, queryReferenced, aggregate, } = DataOperation
+const { eq, ne, string_contains, string_begins, string_ends, gt, gte, lt, lte, include } = AdapterOperators
+const UnsupportedCapabilities = [DataOperation.insertReferences, DataOperation.removeReferences, DataOperation.queryReferenced]
+
+
+export const ReadWriteOperations = Object.values(DataOperation).filter(op => !UnsupportedCapabilities.includes(op))
+export const ReadOnlyOperations = [query, count, queryReferenced, aggregate]
+export const FieldTypes = Object.values(FieldType)
+export const CollectionOperations = Object.values(CollectionOperation)
+export const ColumnsCapabilities = {
+    text: { sortable: true, columnQueryOperators: [eq, ne, string_contains, string_begins, string_ends, include, gt, gte, lt, lte] },
+    url: { sortable: true, columnQueryOperators: [eq, ne, string_contains, string_begins, string_ends, include, gt, gte, lt, lte] },
+    number: { sortable: true, columnQueryOperators: [eq, ne, gt, gte, lt, lte, include] },
+    boolean: { sortable: true, columnQueryOperators: [eq] },
+    image: { sortable: false, columnQueryOperators: [] },
+    object: { sortable: false, columnQueryOperators: [eq, ne, string_contains, string_begins, string_ends, include, gt, gte, lt, lte] },
+    datetime: { sortable: true, columnQueryOperators: [eq, ne, gt, gte, lt, lte] },
+}

libs/external-db-spanner/src/spanner_data_provider.ts

@@ -1,6 +1,6 @@
 import { recordSetToObj, escapeId, patchFieldName, unpatchFieldName, patchFloat, extractFloatFields } from './spanner_utils'
 import { translateErrorCodes } from './sql_exception_translator'
-import { IDataProvider, AdapterFilter as Filter, AdapterAggregation as Aggregation, Item } from '@wix-velo/velo-external-db-types'
+import { IDataProvider, AdapterFilter as Filter, AdapterAggregation as Aggregation, Item, Sort } from '@wix-velo/velo-external-db-types'
 import { Database as SpannerDb } from '@google-cloud/spanner'
 import FilterParser from './sql_filter_transformer'
 
@@ -45,13 +45,14 @@ export default class DataProvider implements IDataProvider {
         return objs[0]['num']
     }
 
-    async insert(collectionName: string, items: Item[], fields: any): Promise <number> {
+    async insert(collectionName: string, items: Item[], fields: any, upsert = false): Promise <number> { 
         const floatFields = extractFloatFields(fields)
-        await this.database.table(collectionName)
-                            .insert(
-                                (items.map((item: any) => patchFloat(item, floatFields)))
-                                        .map(this.asDBEntity.bind(this))
-                            ).catch(translateErrorCodes)
+
+        const preparedItems = items.map((item: any) => patchFloat(item, floatFields)).map(this.asDBEntity.bind(this))
+
+        upsert ? await this.database.table(collectionName).upsert(preparedItems).catch((err) => translateErrorCodes(err, collectionName)) :
+                 await this.database.table(collectionName).insert(preparedItems).catch((err) => translateErrorCodes(err, collectionName))
+
         return items.length
     }
 
@@ -86,7 +87,7 @@ export default class DataProvider implements IDataProvider {
                            .update(
                                (items.map((item: any) => patchFloat(item, floatFields)))
                                      .map(this.asDBEntity.bind(this))
-                           )
+                           ).catch((err) => translateErrorCodes(err, collectionName))
         return items.length
     }
 
@@ -112,13 +113,14 @@ export default class DataProvider implements IDataProvider {
         await this.delete(collectionName, itemIds)
     }
 
-    async aggregate(collectionName: string, filter: Filter, aggregation: Aggregation): Promise <Item[]> {
+    async aggregate(collectionName: string, filter: Filter, aggregation: Aggregation, sort: Sort[], skip: any, limit: any,): Promise <Item[]> {
         const { filterExpr: whereFilterExpr, parameters: whereParameters } = this.filterParser.transform(filter)
+        const { sortExpr } = this.filterParser.orderBy(sort)
 
         const { fieldsStatement, groupByColumns, havingFilter, parameters } = this.filterParser.parseAggregation(aggregation)
         const query = {
-            sql: `SELECT ${fieldsStatement} FROM ${escapeId(collectionName)} ${whereFilterExpr} GROUP BY ${groupByColumns.map(column => escapeId(column)).join(', ')} ${havingFilter}`,
-            params: { ...whereParameters, ...parameters },
+            sql: `SELECT ${fieldsStatement} FROM ${escapeId(collectionName)} ${whereFilterExpr} GROUP BY ${groupByColumns.map(column => escapeId(column)).join(', ')} ${havingFilter} ${sortExpr} LIMIT @limit OFFSET @skip`,
+            params: { ...whereParameters, ...parameters, skip, limit },
         }
 
         const [rows] = await this.database.run(query)

libs/external-db-spanner/src/spanner_schema_provider.ts

@@ -1,10 +1,11 @@
-import { SystemFields, validateSystemFields, parseTableData, AllSchemaOperations } from '@wix-velo/velo-external-db-commons'
+import { SystemFields, validateSystemFields, parseTableData, AllSchemaOperations, EmptyCapabilities } from '@wix-velo/velo-external-db-commons'
 import { errors } from '@wix-velo/velo-external-db-commons'
 import SchemaColumnTranslator from './sql_schema_translator'
 import { notThrowingTranslateErrorCodes } from './sql_exception_translator'
 import { recordSetToObj, escapeId, patchFieldName, unpatchFieldName, escapeFieldId } from './spanner_utils'
 import { Database as SpannerDb } from '@google-cloud/spanner'
-import { InputField, ISchemaProvider, ResponseField, SchemaOperations, Table } from '@wix-velo/velo-external-db-types'
+import { CollectionCapabilities, Encryption, InputField, ISchemaProvider, SchemaOperations, Table } from '@wix-velo/velo-external-db-types'
+import { CollectionOperations, ColumnsCapabilities, FieldTypes, ReadOnlyOperations, ReadWriteOperations } from './spanner_capabilities'
 const { CollectionDoesNotExists, CollectionAlreadyExists } = errors
 
 export default class SchemaProvider implements ISchemaProvider {
@@ -18,22 +19,23 @@ export default class SchemaProvider implements ISchemaProvider {
 
     async list(): Promise<Table[]> {
         const query = {
-            sql: 'SELECT table_name, COLUMN_NAME, SPANNER_TYPE FROM information_schema.columns WHERE table_catalog = @tableCatalog and table_schema = @tableSchema',
+            sql: 'SELECT table_name, COLUMN_NAME as field, SPANNER_TYPE as type FROM information_schema.columns WHERE table_catalog = @tableCatalog and table_schema = @tableSchema',
             params: {
                 tableSchema: '',
                 tableCatalog: '',
             },
         }
 
         const [rows] = await this.database.run(query)
-        const res = recordSetToObj(rows)
+        const res = recordSetToObj(rows) as { table_name: string, field: string, type: string }[]
 
-        const tables: {[x:string]: {table_name: string, field: string, type: string}[]} = parseTableData(res)
+        const tables = parseTableData(res)
 
         return Object.entries(tables)
                      .map(([collectionName, rs]) => ({
                          id: collectionName,
-                         fields: rs.map( this.reformatFields.bind(this) )
+                         fields: rs.map( this.appendAdditionalFieldDetails.bind(this) ),
+                         capabilities: this.collectionCapabilities(rs.map(r => r.field))
                      }))
     }
 
@@ -60,24 +62,24 @@ export default class SchemaProvider implements ISchemaProvider {
                                                                   .join(', ')
         const primaryKeySql = SystemFields.filter(f => f.isPrimary).map(f => escapeFieldId(f.name)).join(', ')
 
-        await this.updateSchema(`CREATE TABLE ${escapeId(collectionName)} (${dbColumnsSql}) PRIMARY KEY (${primaryKeySql})`, CollectionAlreadyExists)
+        await this.updateSchema(`CREATE TABLE ${escapeId(collectionName)} (${dbColumnsSql}) PRIMARY KEY (${primaryKeySql})`, collectionName, CollectionAlreadyExists)
     }
 
     async addColumn(collectionName: string, column: InputField): Promise<void> {
         await validateSystemFields(column.name)
 
-        await this.updateSchema(`ALTER TABLE ${escapeId(collectionName)} ADD COLUMN ${this.sqlSchemaTranslator.columnToDbColumnSql(column)}`)
+        await this.updateSchema(`ALTER TABLE ${escapeId(collectionName)} ADD COLUMN ${this.sqlSchemaTranslator.columnToDbColumnSql(column)}`, collectionName)
     }
 
     async removeColumn(collectionName: string, columnName: string): Promise<void> {
         await validateSystemFields(columnName)
 
-        await this.updateSchema(`ALTER TABLE ${escapeId(collectionName)} DROP COLUMN ${escapeId(columnName)}`)
+        await this.updateSchema(`ALTER TABLE ${escapeId(collectionName)} DROP COLUMN ${escapeId(columnName)}`, collectionName)
     }
 
-    async describeCollection(collectionName: string): Promise<ResponseField[]> {
+    async describeCollection(collectionName: string): Promise<Table> {
         const query = {
-            sql: 'SELECT table_name, COLUMN_NAME, SPANNER_TYPE FROM information_schema.columns WHERE table_catalog = @tableCatalog and table_schema = @tableSchema and table_name = @tableName',
+            sql: 'SELECT table_name, COLUMN_NAME as field, SPANNER_TYPE as type FROM information_schema.columns WHERE table_catalog = @tableCatalog and table_schema = @tableSchema and table_name = @tableName',
             params: {
                 tableSchema: '',
                 tableCatalog: '',
@@ -89,40 +91,58 @@ export default class SchemaProvider implements ISchemaProvider {
         const res = recordSetToObj(rows)
 
         if (res.length === 0) {
-            throw new CollectionDoesNotExists('Collection does not exists')
+            throw new CollectionDoesNotExists('Collection does not exists', collectionName)
         }
 
-        return res.map( this.reformatFields.bind(this) )
+        return {
+            id: collectionName,
+            fields: res.map( this.appendAdditionalFieldDetails.bind(this) ),
+            capabilities: this.collectionCapabilities(res.map(f => f.field))
+        }
     }
 
     async drop(collectionName: string): Promise<void> {
-        await this.updateSchema(`DROP TABLE ${escapeId(collectionName)}`)
+        await this.updateSchema(`DROP TABLE ${escapeId(collectionName)}`, collectionName)
     }
 
+    async changeColumnType(collectionName: string, _column: InputField): Promise<void> {
+        throw new errors.UnsupportedSchemaOperation('changeColumnType is not supported', collectionName, 'changeColumnType')
+    }
 
-    async updateSchema(sql: string, catching: any = undefined) {
+    async updateSchema(sql: string, collectionName: string, catching: any = undefined ) {
         try {
             const [operation] = await this.database.updateSchema([sql])
 
             await operation.promise()
         } catch (err) {
-            const e = notThrowingTranslateErrorCodes(err)
+            const e = notThrowingTranslateErrorCodes(err, collectionName)
             if (!catching || (catching && !(e instanceof catching))) {
                 throw e
             }
         }
     }
 
-    fixColumn(c: InputField) {
+    private fixColumn(c: InputField) {
         return { ...c, name: patchFieldName(c.name) }
     }
 
-    reformatFields(r: { [x: string]: string }) {
-        const { type, subtype } = this.sqlSchemaTranslator.translateType(r['SPANNER_TYPE'])
+    private appendAdditionalFieldDetails(row: { field: string, type: string }) {
+        const type = this.sqlSchemaTranslator.translateType(row.type).type as keyof typeof ColumnsCapabilities
+        return {
+            field: unpatchFieldName(row.field),
+            ...this.sqlSchemaTranslator.translateType(row.type),
+            capabilities: ColumnsCapabilities[type] ?? EmptyCapabilities
+        }
+    }
+
+    private collectionCapabilities(fieldNames: string[]): CollectionCapabilities {
         return {
-            field: unpatchFieldName(r['COLUMN_NAME']),
-            type,
-            subtype
+            dataOperations: fieldNames.map(unpatchFieldName).includes('_id') ? ReadWriteOperations : ReadOnlyOperations,
+            fieldTypes: FieldTypes,
+            collectionOperations: CollectionOperations,
+            referenceCapabilities: { supportedNamespaces: [] },
+            indexing: [],
+            encryption: Encryption.notSupported
         }
     }
 

libs/external-db-spanner/src/sql_exception_translator.ts

@@ -1,7 +1,16 @@
 import { errors } from '@wix-velo/velo-external-db-commons'
 const { CollectionDoesNotExists, FieldAlreadyExists, FieldDoesNotExist, DbConnectionError, CollectionAlreadyExists, ItemAlreadyExists, InvalidQuery, UnrecognizedError } = errors
+const extractId = (msg: string | null) => {
+    msg = msg || ''
+    const regex = /String\("([A-Za-z0-9-]+)"\)/i
+    const match = msg.match(regex)
+    if (match) {
+      return match[1]
+    }
+    return ''
+  }
 
-export const notThrowingTranslateErrorCodes = (err: any) => {
+export const notThrowingTranslateErrorCodes = (err: any, collectionName?: string) => {
     switch (err.code) {
         case 9:
             if (err.details.includes('column')) {
@@ -11,19 +20,22 @@ export const notThrowingTranslateErrorCodes = (err: any) => {
             }
         case 5:
             if (err.details.includes('Column')) {
-                return new FieldDoesNotExist(err.details)
+                return new FieldDoesNotExist(err.details, collectionName)
             } else if (err.details.includes('Instance')) {
                 return new DbConnectionError(`Access to database denied - wrong credentials or host is unavailable, sql message:  ${err.details} `)
             } else if (err.details.includes('Database')) {
                 return new DbConnectionError(`Database does not exists or you don't have access to it, sql message: ${err.details}`)
             } else if (err.details.includes('Table')) {
-                return new CollectionDoesNotExists(err.details)
+                console.log({ details: err.details, collectionName })
+
+                return new CollectionDoesNotExists(err.details, collectionName)
             } else {
                 return new InvalidQuery(`${err.details}`)
             }
         case 6:
             if (err.details.includes('already exists')) 
-                return new ItemAlreadyExists(`Item already exists: ${err.details}`)
+                return new ItemAlreadyExists(`Item already exists: ${err.details}`, collectionName, extractId(err.details))
+
             else
                 return new InvalidQuery(`${err.details}`)
         case 7:
@@ -35,6 +47,6 @@ export const notThrowingTranslateErrorCodes = (err: any) => {
     }
 }
 
-export const translateErrorCodes = (err: any) => {
-    throw notThrowingTranslateErrorCodes(err)
+export const translateErrorCodes = (err: any, collectionName?: string) => {
+    throw notThrowingTranslateErrorCodes(err, collectionName)
 }