Skip to content

Commit

Permalink
issues supabase#145 and supabase#149 fixed
Browse files Browse the repository at this point in the history
  • Loading branch information
@kamysh
kamysh committed Feb 4, 2023
1 parent 102f806 commit d5b7594
Show file tree
Hide file tree
Showing 5 changed files with 100 additions and 32 deletions.
5 changes: 2 additions & 3 deletions DbAnalyzer/go/oracle.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@ import (
func getOraInfo(c types.Connection) (interface{}, error) {

query := url.Values{}
query.Add("database", c.Database)
if c.Tls {
query.Add("SSL", "true")
} else {
Expand All @@ -24,7 +23,7 @@ func getOraInfo(c types.Connection) (interface{}, error) {
Scheme: "oracle",
User: url.UserPassword(c.User, c.Password),
Host: fmt.Sprintf("%s:%d", c.Address, c.Port),
Path: "XEPDB1",
Path: strings.ToUpper(c.Database),
RawQuery: query.Encode(),
}
oracleconn := u.String()
Expand Down Expand Up @@ -143,7 +142,7 @@ func getOraInfo(c types.Connection) (interface{}, error) {
}
if curSchema == -1 || schema != database.Schemas[curSchema].Name {
switch schema {
case "information_schema", "mysql", "sys", "performance_schema":
case "RDSADMIN":
isSystem = true
default:
isSystem = false
Expand Down
40 changes: 39 additions & 1 deletion DbGuardian/foreign_data_wrappers/mysql_fdw/mysql_fdw.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -664,9 +664,9 @@ mysqlBeginForeignScan(ForeignScanState *node, int eflags)
options = mysql_get_options(rte->relid, true);

/* Dymium */
festate->how_to_redact = (int*)palloc0(sizeof(int) * tupleDescriptor->natts);
if (fsplan->scan.scanrelid > 0)
{
festate->how_to_redact = (int*)palloc0(sizeof(int) * tupleDescriptor->natts);
/* !Dymium! */
for(int k = 0; k != tupleDescriptor->natts; ++k) {
List *options = GetForeignColumnOptions(rte->relid, TupleDescAttr(tupleDescriptor, k)->attnum);
Expand All @@ -684,6 +684,44 @@ mysqlBeginForeignScan(ForeignScanState *node, int eflags)
}
}
}
else
{
for(int k = 0; k != tupleDescriptor->natts; ++k) {
Form_pg_attribute att = TupleDescAttr(tupleDescriptor, k);
Var *var;
RangeTblEntry *rte;
Oid reltype;
List *options;
ListCell *lc;

/*
* If we can't identify the referenced table, do nothing. This'll
* likely lead to failure later, but perhaps we can muddle through.
*/
var = (Var *) list_nth_node(TargetEntry, fsplan->fdw_scan_tlist,
k)->expr;
if (!IsA(var, Var) /* !Dymium: move this condition down: || var->varattno != 0 */)
continue;
rte = list_nth(estate->es_range_table, var->varno - 1);
if (rte->rtekind != RTE_RELATION)
continue;
reltype = get_rel_type_id(rte->relid);
if (!OidIsValid(reltype))
continue;

options = GetForeignColumnOptions(rte->relid, att->attnum);
foreach(lc, options)
{
DefElem *def = (DefElem *) lfirst(lc);
if (strcmp(def->defname, "redact") == 0)
{
char *redact = defGetString(def);
festate->how_to_redact[k] = atoi(redact);
break;
}
}
}
}

/*
* Get the already connected connection, otherwise connect and get the
Expand Down
47 changes: 39 additions & 8 deletions DbGuardian/foreign_data_wrappers/postgres_fdw/postgres_fdw.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1437,7 +1437,7 @@ postgresGetForeignPlan(PlannerInfo *root,
* Construct a tuple descriptor for the scan tuples handled by a foreign join.
*/
static TupleDesc
get_tupdesc_for_join_scan_tuples(ForeignScanState *node)
get_tupdesc_for_join_scan_tuples(ForeignScanState *node, int **how_to_redact /* !Dymium */)
{
ForeignScan *fsplan = (ForeignScan *) node->ss.ps.plan;
EState *estate = node->ss.ps.state;
Expand All @@ -1453,31 +1453,56 @@ get_tupdesc_for_join_scan_tuples(ForeignScanState *node)
* we can convert them to a composite type the local server knows.
*/
tupdesc = CreateTupleDescCopy(node->ss.ss_ScanTupleSlot->tts_tupleDescriptor);

*how_to_redact = (int*)palloc0(sizeof(int) * tupdesc->natts);

for (int i = 0; i < tupdesc->natts; i++)
{
Form_pg_attribute att = TupleDescAttr(tupdesc, i);
Var *var;
RangeTblEntry *rte;
Oid reltype;

/* Nothing to do if it's not a generic RECORD attribute */
if (att->atttypid != RECORDOID || att->atttypmod >= 0)
continue;
List *options;
ListCell *lc;

/*
* If we can't identify the referenced table, do nothing. This'll
* likely lead to failure later, but perhaps we can muddle through.
*/
var = (Var *) list_nth_node(TargetEntry, fsplan->fdw_scan_tlist,
i)->expr;
if (!IsA(var, Var) || var->varattno != 0)
if (!IsA(var, Var) /* !Dymium: move this condition down: || var->varattno != 0 */)
continue;
rte = list_nth(estate->es_range_table, var->varno - 1);
if (rte->rtekind != RTE_RELATION)
continue;
reltype = get_rel_type_id(rte->relid);
if (!OidIsValid(reltype))
continue;


/* !Dymium: find how_to_redact */
options = GetForeignColumnOptions(rte->relid, att->attnum);
foreach(lc, options)
{
DefElem *def = (DefElem *) lfirst(lc);
if (strcmp(def->defname, "redact") == 0)
{
char *redact = defGetString(def);
(*how_to_redact)[i] = atoi(redact);
break;
}
}

/* !Dymium: move these conditions here */
/* Nothing to do if it's not a generic RECORD attribute */
if (att->atttypid != RECORDOID || att->atttypmod >= 0)
continue;

if (var->varattno != 0)
continue;

att->atttypid = reltype;
/* shouldn't need to change anything else */
}
Expand Down Expand Up @@ -1508,7 +1533,9 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags)
return;

// Dymium
// puts("===== FSPLAN START =====");
// pprint(fsplan);
// puts("===== FSPLAN END =====");

/*
* We'll save private state in node->fdw_state.
Expand Down Expand Up @@ -1546,6 +1573,7 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags)
/* Get private info created by planner functions. */
fsstate->query = strVal(list_nth(fsplan->fdw_private,
FdwScanPrivateSelectSql));
// puts(fsstate->query);
fsstate->retrieved_attrs = (List *) list_nth(fsplan->fdw_private,
FdwScanPrivateRetrievedAttrs);
fsstate->fetch_size = intVal(list_nth(fsplan->fdw_private,
Expand Down Expand Up @@ -1592,7 +1620,7 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags)
else
{
fsstate->rel = NULL;
fsstate->tupdesc = get_tupdesc_for_join_scan_tuples(node);
fsstate->tupdesc = get_tupdesc_for_join_scan_tuples(node, &fsstate->how_to_redact);
}

fsstate->attinmeta = TupleDescGetAttInMetadata(fsstate->tupdesc);
Expand Down Expand Up @@ -2730,8 +2758,11 @@ postgresBeginDirectModify(ForeignScanState *node, int eflags)
{
TupleDesc tupdesc;

if (fsplan->scan.scanrelid == 0)
tupdesc = get_tupdesc_for_join_scan_tuples(node);
if (fsplan->scan.scanrelid == 0) {
/* !Dymium: not used */
int *how_to_redact;
tupdesc = get_tupdesc_for_join_scan_tuples(node, &how_to_redact);
}
else
tupdesc = RelationGetDescr(dmstate->rel);

Expand Down
20 changes: 10 additions & 10 deletions DbGuardian/go/DbSetup.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -87,12 +87,12 @@ func options(connectionType types.ConnectionType) iOptions {
host, port, strings.ToUpper(dbname))
},
userMapping: func(user, password string) string {
return fmt.Sprintf("username '%s', password '%s'",
return fmt.Sprintf("user '%s', password '%s'",
esc(user), esc(password))
},
table: func(remoteSchema, remoteTable string) string {
return fmt.Sprintf("schema '%s', table '%s'",
esc(remoteSchema), esc(strings.ToUpper(remoteTable)))
esc(remoteSchema), esc(remoteTable))
},
}
}
Expand Down Expand Up @@ -218,21 +218,21 @@ func configureDatabase(db *sql.DB,
}

for k := range shortSchemas {
if err := exec("CREATE SCHEMA IF NOT EXISTS " + k); err != nil {
if err := exec("CREATE SCHEMA IF NOT EXISTS " + strings.ToLower(k)); err != nil {
return err
}
if err := exec("GRANT USAGE ON SCHEMA " + k + " TO " + localUser); err != nil {
if err := exec("GRANT USAGE ON SCHEMA " + strings.ToLower(k) + " TO " + localUser); err != nil {
return err
}
if err := exec("ALTER DEFAULT PRIVILEGES IN SCHEMA " + k + " GRANT SELECT ON TABLES TO " + localUser); err != nil {
if err := exec("ALTER DEFAULT PRIVILEGES IN SCHEMA " + strings.ToLower(k) + " GRANT SELECT ON TABLES TO " + localUser); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, "INSERT INTO _dymium.schemas (\"schema\") VALUES ( $1 )", k); err != nil {
return rollback(err, "Registering schema "+k+"_server failed")
if _, err := tx.ExecContext(ctx, "INSERT INTO _dymium.schemas (\"schema\") VALUES ( $1 )", strings.ToLower(k)); err != nil {
return rollback(err, "Registering schema "+strings.ToLower(k)+"_server failed")
}
}
for k := range longSchemas {
kk := k.k1 + "_" + k.k2
kk := strings.ToLower(k.k1 + "_" + k.k2)
if err := exec(fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS %q", kk)); err != nil {
return err
}
Expand Down Expand Up @@ -341,7 +341,7 @@ func configureDatabase(db *sql.DB,
rtyp = uuid
}
defs = append(defs, fmt.Sprintf(" %q %s OPTIONS( redact '%d' )%s",
c.Name, c.Typ, int(ract)|(rnul<<2)|(int(rtyp)<<3), notNull))
strings.ToLower(c.Name), c.Typ, int(ract)|(rnul<<2)|(int(rtyp)<<3), notNull))
}
}
e := "CREATE FOREIGN TABLE %q.%q (\n" + strings.Join(defs, ",\n") + "\n)\n" +
Expand All @@ -358,7 +358,7 @@ func configureDatabase(db *sql.DB,
}
}
for _, sch := range schs {
if err := exec(fmt.Sprintf(e, sch, t.Name, con.Name, opts.table(s.Name, t.Name))); err != nil {
if err := exec(fmt.Sprintf(e, strings.ToLower(sch), strings.ToLower(t.Name), con.Name, opts.table(s.Name, t.Name))); err != nil {
return err
}
}
Expand Down
20 changes: 10 additions & 10 deletions DbSync/go/DbSetup.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -87,12 +87,12 @@ func options(connectionType types.ConnectionType) iOptions {
host, port, strings.ToUpper(dbname))
},
userMapping: func(user, password string) string {
return fmt.Sprintf("username '%s', password '%s'",
return fmt.Sprintf("user '%s', password '%s'",
esc(user), esc(password))
},
table: func(remoteSchema, remoteTable string) string {
return fmt.Sprintf("schema '%s', table '%s'",
esc(remoteSchema), esc(strings.ToUpper(remoteTable)))
esc(remoteSchema), esc(remoteTable))
},
}
}
Expand Down Expand Up @@ -218,21 +218,21 @@ func configureDatabase(db *sql.DB,
}

for k := range shortSchemas {
if err := exec("CREATE SCHEMA IF NOT EXISTS " + k); err != nil {
if err := exec("CREATE SCHEMA IF NOT EXISTS " + strings.ToLower(k)); err != nil {
return err
}
if err := exec("GRANT USAGE ON SCHEMA " + k + " TO " + localUser); err != nil {
if err := exec("GRANT USAGE ON SCHEMA " + strings.ToLower(k) + " TO " + localUser); err != nil {
return err
}
if err := exec("ALTER DEFAULT PRIVILEGES IN SCHEMA " + k + " GRANT SELECT ON TABLES TO " + localUser); err != nil {
if err := exec("ALTER DEFAULT PRIVILEGES IN SCHEMA " + strings.ToLower(k) + " GRANT SELECT ON TABLES TO " + localUser); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, "INSERT INTO _dymium.schemas (\"schema\") VALUES ( $1 )", k); err != nil {
return rollback(err, "Registering schema "+k+"_server failed")
if _, err := tx.ExecContext(ctx, "INSERT INTO _dymium.schemas (\"schema\") VALUES ( $1 )", strings.ToLower(k)); err != nil {
return rollback(err, "Registering schema "+strings.ToLower(k)+"_server failed")
}
}
for k := range longSchemas {
kk := k.k1 + "_" + k.k2
kk := strings.ToLower(k.k1 + "_" + k.k2)
if err := exec(fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS %q", kk)); err != nil {
return err
}
Expand Down Expand Up @@ -341,7 +341,7 @@ func configureDatabase(db *sql.DB,
rtyp = uuid
}
defs = append(defs, fmt.Sprintf(" %q %s OPTIONS( redact '%d' )%s",
c.Name, c.Typ, int(ract)|(rnul<<2)|(int(rtyp)<<3), notNull))
strings.ToLower(c.Name), c.Typ, int(ract)|(rnul<<2)|(int(rtyp)<<3), notNull))
}
}
e := "CREATE FOREIGN TABLE %q.%q (\n" + strings.Join(defs, ",\n") + "\n)\n" +
Expand All @@ -358,7 +358,7 @@ func configureDatabase(db *sql.DB,
}
}
for _, sch := range schs {
if err := exec(fmt.Sprintf(e, sch, t.Name, con.Name, opts.table(s.Name, t.Name))); err != nil {
if err := exec(fmt.Sprintf(e, strings.ToLower(sch), strings.ToLower(t.Name), con.Name, opts.table(s.Name, t.Name))); err != nil {
return err
}
}
Expand Down

0 comments on commit d5b7594

Please sign in to comment.