Release v1.7.0

Author: exAspArk

README.md

@@ -317,6 +317,7 @@ SELECT * FROM [TABLE] WHERE [JSON_COLUMN]->>'[JSON_KEY]' = '[JSON_VALUE]';
   - [x] DBeaver
   - [x] pgAdmin
   - [x] Grafana
+  - [x] Retool
   - [ ] Jupyter notebooks ([#27](https://github.com/BemiHQ/BemiDB/issues/27))
 - [x] Data syncing from other sources
   - [x] Amplitude (incremental)

src/common/common_config.go

@@ -1,7 +1,7 @@
 package common
 
 const (
-	VERSION = "1.6.0"
+	VERSION = "1.7.0"
 
 	ENV_LOG_LEVEL                   = "BEMIDB_LOG_LEVEL"
 	ENV_DISABLE_ANONYMOUS_ANALYTICS = "BEMIDB_DISABLE_ANONYMOUS_ANALYTICS"

src/server/parser_function.go

@@ -31,7 +31,7 @@ func (parser *ParserFunction) FirstArgumentToString(functionCall *pgQuery.FuncCa
 // n from (FUNCTION()).n
 func (parser *ParserFunction) IndirectionName(targetNode *pgQuery.Node) string {
 	indirection := targetNode.GetResTarget().Val.GetAIndirection()
-	if indirection != nil {
+	if indirection != nil && len(indirection.Indirection) == 1 && indirection.Indirection[0].GetString_() != nil {
 		return indirection.Indirection[0].GetString_().Sval
 	}
 

src/server/parser_table.go

@@ -55,7 +55,11 @@ func (parser *ParserTable) MakeIcebergTableNode(queryToIcebergTable QueryToIcebe
 	if permissions == nil {
 		query = "SELECT * FROM iceberg_scan('" + queryToIcebergTable.IcebergTablePath + "')"
 	} else if columnNames, allowed := (*permissions)[queryToIcebergTable.QuerySchemaTable.ToIcebergSchemaTable().ToArg()]; allowed {
-		query = "SELECT " + strings.Join(columnNames, ", ") + " FROM iceberg_scan('" + queryToIcebergTable.IcebergTablePath + "')"
+		quotedColumnNames := make([]string, len(columnNames))
+		for i, columnName := range columnNames {
+			quotedColumnNames[i] = "\"" + columnName + "\""
+		}
+		query = "SELECT " + strings.Join(quotedColumnNames, ", ") + " FROM iceberg_scan('" + queryToIcebergTable.IcebergTablePath + "')"
 	} else {
 		query = "SELECT NULL WHERE FALSE"
 	}

src/server/postgres_server.go

@@ -174,7 +174,13 @@ func (server *PostgresServer) handleExtendedQuery(queryHandler *QueryHandler, pa
 		case *pgproto3.Parse:
 			server.handleExtendedQuery(queryHandler, message) // self-recursion
 			return nil
+		case *pgproto3.Flush:
+			// Ignore Flush messages, as we are sending responses immediately.
+		case *pgproto3.Close:
+			common.LogDebug(server.config.CommonConfig, "Closing prepared statement", message.Name)
+			server.writeMessages(&pgproto3.CloseComplete{})
 		default:
+			common.LogError(server.config.CommonConfig, fmt.Sprintf("Received unexpected message type from client: %T", message))
 			return fmt.Errorf("received unexpected message type from client: %T", message)
 		}
 	}

src/server/query_handler_test.go

@@ -83,6 +83,11 @@ func TestHandleQuery(t *testing.T) {
 				"types":       {uint32ToString(pgtype.Int4OID)},
 				"values":      {"0"},
 			},
+			"SELECT pg_cancel_backend(12345) AS pg_cancel_backend": {
+				"description": {"pg_cancel_backend"},
+				"types":       {uint32ToString(pgtype.BoolOID)},
+				"values":      {"t"},
+			},
 			"SELECT * from pg_is_in_recovery()": {
 				"description": {"pg_is_in_recovery"},
 				"types":       {uint32ToString(pgtype.BoolOID)},
@@ -400,6 +405,11 @@ func TestHandleQuery(t *testing.T) {
 				"types":       {uint32ToString(pgtype.OIDOID)},
 				"values":      {"25"},
 			},
+			"SELECT DISTINCT ON(typlen) oid FROM pg_type ORDER BY oid LIMIT 1": {
+				"description": {"oid"},
+				"types":       {uint32ToString(pgtype.OIDOID)},
+				"values":      {"16"},
+			},
 		})
 	})
 

src/server/query_remapper.go

@@ -509,6 +509,12 @@ func (remapper *QueryRemapper) remapSelect(selectStatement *pgQuery.SelectStmt,
 		}
 	}
 
+	// DISTINCT ON (column)
+	distinctClauses := selectStatement.GetDistinctClause()
+	for i, distinctNode := range distinctClauses {
+		distinctClauses[i] = remapper.remappedExpressions(distinctNode, remappedColumnRefs, permissions, indentLevel) // recursion
+	}
+
 	return remappedColumnRefs
 }
 

src/server/query_remapper_function.go

@@ -17,6 +17,7 @@ func CreatePgCatalogMacroQueries(config *Config) []string {
 		"CREATE MACRO aclexplode(aclitem_array) AS json(aclitem_array)",
 		"CREATE MACRO current_setting(setting_name) AS '', (setting_name, missing_ok) AS ''",
 		"CREATE MACRO pg_backend_pid() AS 0",
+		"CREATE MACRO pg_cancel_backend(pid) AS true",
 		"CREATE MACRO pg_encoding_to_char(encoding_int) AS 'UTF8'",
 		"CREATE MACRO pg_get_expr(pg_node_tree, relation_oid) AS pg_catalog.pg_get_expr(pg_node_tree, relation_oid), (pg_node_tree, relation_oid, pretty_bool) AS pg_catalog.pg_get_expr(pg_node_tree, relation_oid)",
 		"CREATE MACRO pg_get_function_identity_arguments(func_oid) AS ''",