feat(redshift): support external relations (#108)

# Description Adding support for external schemas for the following db methods: - `ListSchemas` - `SchemaExists` - `ListTables` - `ListTablesWithPrefix` - `TableExists` - `ListColumns` ## Linear Ticket resolves REV-1064 ## Security - [x] The code changed/added as part of this pull request won't create any security issues with how the software is being used. --------- Co-authored-by: Vikas Venkatraman <[email protected]>
rudderlabs · Jun 20, 2024 · 737f1ed · 737f1ed
1 parent 8f62563
commit 737f1ed
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 5 deletions.
diff --git a/sqlconnect/internal/redshift/db.go b/sqlconnect/internal/redshift/db.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	_ "github.com/lib/pq" // postgres driver
+	"github.com/samber/lo"
 	"github.com/tidwall/gjson"
 
 	redshiftdriver "github.com/rudderlabs/sqlconnect-go/sqlconnect/internal/redshift/driver"
@@ -49,15 +50,28 @@ func NewDB(credentialsJSON json.RawMessage) (*DB, error) {
 					return "SELECT current_database()"
 				}
 				cmds.ListSchemas = func() (string, string) {
-					return "SELECT schema_name FROM svv_redshift_schemas", "schema_name"
+					return "SELECT schema_name FROM svv_all_schemas", "schema_name"
 				}
 				cmds.SchemaExists = func(schema base.UnquotedIdentifier) string {
-					return fmt.Sprintf("SELECT schema_name FROM svv_redshift_schemas WHERE schema_name = '%[1]s'", schema)
+					return fmt.Sprintf("SELECT schema_name FROM svv_all_schemas WHERE schema_name = '%[1]s'", schema)
+				}
+				cmds.ListTables = func(schema base.UnquotedIdentifier) (sqlAndColumnNamePairs []lo.Tuple2[string, string]) {
+					return []lo.Tuple2[string, string]{
+						{A: fmt.Sprintf("SELECT table_name FROM svv_all_tables WHERE schema_name = '%[1]s'", schema), B: "table_name"},
+					}
+				}
+				cmds.ListTablesWithPrefix = func(schema base.UnquotedIdentifier, prefix string) []lo.Tuple2[string, string] {
+					return []lo.Tuple2[string, string]{
+						{A: fmt.Sprintf("SELECT table_name FROM svv_all_tables WHERE schema_name='%[1]s' AND table_name LIKE '%[2]s'", schema, prefix+"%"), B: "table_name"},
+					}
+				}
+				cmds.TableExists = func(schema, table base.UnquotedIdentifier) string {
+					return fmt.Sprintf("SELECT table_name FROM svv_all_tables WHERE schema_name='%[1]s' and table_name = '%[2]s'", schema, table)
 				}
 				cmds.ListColumns = func(catalog, schema, table base.UnquotedIdentifier) (string, string, string) {
-					stmt := fmt.Sprintf("SELECT column_name, data_type FROM SVV_COLUMNS WHERE table_schema = '%[1]s' AND table_name = '%[2]s'", schema, table)
+					stmt := fmt.Sprintf("SELECT column_name, data_type FROM SVV_ALL_COLUMNS WHERE schema_name = '%[1]s' AND table_name = '%[2]s'", schema, table)
 					if catalog != "" {
-						stmt += fmt.Sprintf(" AND table_catalog = '%[1]s'", catalog)
+						stmt += fmt.Sprintf(" AND database_name = '%[1]s'", catalog)
 					}
 					return stmt + " ORDER BY ordinal_position ASC", "column_name", "data_type"
 				}

diff --git a/sqlconnect/internal/redshift/testdata/legacy-column-mapping-test-columns-table.json b/sqlconnect/internal/redshift/testdata/legacy-column-mapping-test-columns-table.json
@@ -21,7 +21,7 @@
     "_character": "string",
     "_date": "datetime",
     "_time": "time without time zone",
-    "_timetz": "timetz",
+    "_timetz": "time with time zone",
     "_timestamptz": "datetime",
     "_timestampntz": "datetime",
     "_timestampwtz": "datetime",