Files
2026-07-13 12:32:25 +08:00

373 lines
12 KiB
Go

// Copyright 2026 Dolthub, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package tables
import (
"fmt"
"strings"
"github.com/dolthub/dolt/go/libraries/doltcore/sqle"
"github.com/dolthub/dolt/go/libraries/doltcore/sqle/resolve"
"github.com/dolthub/go-mysql-server/sql"
"github.com/dolthub/doltgresql/core"
"github.com/dolthub/doltgresql/core/id"
)
// PgDatabase wraps a sqle.Database to add PostgreSQL-specific behavior.
type PgDatabase struct {
sqle.Database
}
var _ sql.DatabaseSchema = &PgDatabase{}
var _ sql.SchemaDatabase = &PgDatabase{}
var _ sql.SchemaObjectNameValidator = &PgDatabase{}
var _ sql.IndexNameGenerator = &PgDatabase{}
// PgReadOnlyDatabase is the read-only variant of PgDatabase, used for revision databases
// such as "postgres/main" returned by sqle.DoltDatabaseProvider for detached-HEAD sessions.
// It applies the same schema-wrapping logic as PgDatabase.
type PgReadOnlyDatabase struct {
sqle.ReadOnlyDatabase
}
var _ sql.DatabaseSchema = &PgReadOnlyDatabase{}
var _ sql.SchemaDatabase = &PgReadOnlyDatabase{}
// WrapSqleDatabase creates a PgDatabase from a sqle.Database.
func WrapSqleDatabase(db sqle.Database) *PgDatabase {
return &PgDatabase{db}
}
// WrapSqlDatabase wraps any Dolt database variant as a Pg-aware database.
// ReadOnlyDatabase embeds sqle.Database by value, so a plain sqle.Database type
// assertion does not match it; this function handles both cases.
func WrapSqlDatabase(db sql.Database) sql.Database {
if rodb, ok := db.(sqle.ReadOnlyDatabase); ok {
return &PgReadOnlyDatabase{rodb}
}
if sdb, ok := db.(sqle.Database); ok {
return WrapSqleDatabase(sdb)
}
return db
}
// applySchemaWrap wraps a single schema returned by the underlying sqle.Database methods.
// System schemas (those with registered virtual-table handlers) get a Database wrapper
// that exposes only virtual tables; all others get a PgDatabase wrapper.
func applySchemaWrap(requestedName string, schema sql.DatabaseSchema) sql.DatabaseSchema {
sdb, ok := schema.(sqle.Database)
if !ok {
// information_schema and any other non-sqle schema: leave as-is.
return schema
}
if _, isSystem := handlers[requestedName]; isSystem {
return Database{sdb}
}
return &PgDatabase{sdb}
}
// AllSchemas overrides sqle.Database.AllSchemas to apply Doltgres schema wrapping.
func (d *PgDatabase) AllSchemas(ctx *sql.Context) ([]sql.DatabaseSchema, error) {
schemas, err := d.Database.AllSchemas(ctx)
if err != nil {
return nil, err
}
for i, s := range schemas {
schemas[i] = applySchemaWrap(s.SchemaName(), s)
}
return schemas, nil
}
// GetSchema overrides sqle.Database.GetSchema to apply Doltgres schema wrapping.
func (d *PgDatabase) GetSchema(ctx *sql.Context, schemaName string) (sql.DatabaseSchema, bool, error) {
schema, ok, err := d.Database.GetSchema(ctx, schemaName)
if !ok || err != nil {
return schema, ok, err
}
return applySchemaWrap(schemaName, schema), true, nil
}
// GetTableInsensitive overrides sqle.Database.GetTableInsensitive to check the pg_catalog
// virtual schema before falling back to user tables.
func (d *PgDatabase) GetTableInsensitive(ctx *sql.Context, tblName string) (sql.Table, bool, error) {
if resolve.UseSearchPath && d.Database.Schema() == "" && strings.HasPrefix(strings.ToLower(tblName), "pg_") {
sdb, found, err := d.GetSchema(ctx, "pg_catalog")
if err != nil {
return nil, false, err
}
if found {
tbl, foundTbl, err := sdb.GetTableInsensitive(ctx, tblName)
if err != nil {
return nil, false, err
}
if foundTbl {
return tbl, true, nil
}
}
}
return d.Database.GetTableInsensitive(ctx, tblName)
}
// DropTable overrides sqle.Database.DropTable to prevent dropping virtual pg_catalog tables.
func (d *PgDatabase) DropTable(ctx *sql.Context, tableName string) error {
if resolve.UseSearchPath && d.Database.Schema() == "" && strings.HasPrefix(strings.ToLower(tableName), "pg_") {
sdb, found, err := d.GetSchema(ctx, "pg_catalog")
if err != nil {
return err
}
if found {
_, foundTbl, err := sdb.GetTableInsensitive(ctx, tableName)
if err != nil {
return err
}
if foundTbl {
return sql.ErrDropTableNotSupported.New("pg_catalog")
}
}
}
return d.Database.DropTable(ctx, tableName)
}
// AllSchemas overrides sqle.ReadOnlyDatabase.AllSchemas to apply Doltgres schema wrapping.
func (d *PgReadOnlyDatabase) AllSchemas(ctx *sql.Context) ([]sql.DatabaseSchema, error) {
schemas, err := d.ReadOnlyDatabase.AllSchemas(ctx)
if err != nil {
return nil, err
}
for i, s := range schemas {
schemas[i] = applySchemaWrap(s.SchemaName(), s)
}
return schemas, nil
}
// GetSchema overrides sqle.ReadOnlyDatabase.GetSchema to apply Doltgres schema wrapping.
func (d *PgReadOnlyDatabase) GetSchema(ctx *sql.Context, schemaName string) (sql.DatabaseSchema, bool, error) {
schema, ok, err := d.ReadOnlyDatabase.GetSchema(ctx, schemaName)
if !ok || err != nil {
return schema, ok, err
}
return applySchemaWrap(schemaName, schema), true, nil
}
// GetTableInsensitive overrides sqle.Database.GetTableInsensitive to check the pg_catalog
// virtual schema before falling back to user tables.
func (d *PgReadOnlyDatabase) GetTableInsensitive(ctx *sql.Context, tblName string) (sql.Table, bool, error) {
if resolve.UseSearchPath && d.ReadOnlyDatabase.Schema() == "" && strings.HasPrefix(strings.ToLower(tblName), "pg_") {
sdb, found, err := d.GetSchema(ctx, "pg_catalog")
if err != nil {
return nil, false, err
}
if found {
tbl, foundTbl, err := sdb.GetTableInsensitive(ctx, tblName)
if err != nil {
return nil, false, err
}
if foundTbl {
return tbl, true, nil
}
}
}
return d.ReadOnlyDatabase.GetTableInsensitive(ctx, tblName)
}
// ValidateNewIndexName implements the sql.SchemaObjectNameValidator interface
func (d *PgDatabase) ValidateNewIndexName(ctx *sql.Context, newIndexName string, skipIfExists bool) (nameAlreadyUsed bool, err error) {
nameAlreadyUsed, _, err = d.doesRelationExist(ctx, newIndexName)
if err != nil {
return false, err
}
if !nameAlreadyUsed {
return false, nil
}
if skipIfExists {
return true, nil
}
return nameAlreadyUsed, fmt.Errorf(`relation "%s" already exists`, newIndexName)
}
// ValidateNewSequenceName implements the sql.SchemaObjectNameValidator interface
func (d *PgDatabase) ValidateNewSequenceName(ctx *sql.Context, newSequenceName string, skipIfExists bool) (nameAlreadyUsed bool, err error) {
nameAlreadyUsed, _, err = d.doesRelationExist(ctx, newSequenceName)
if err != nil {
return false, err
}
if !nameAlreadyUsed {
return false, nil
}
if skipIfExists {
return true, nil
}
return nameAlreadyUsed, fmt.Errorf(`relation "%s" already exists`, newSequenceName)
}
// ValidateNewViewName implements the sql.SchemaObjectNameValidator interface
func (d *PgDatabase) ValidateNewViewName(ctx *sql.Context, newViewName string, replaceAllowed bool) error {
relationExists, relationType, err := d.doesRelationExist(ctx, newViewName)
if err != nil {
return err
}
if !relationExists {
return nil
}
// When REPLACE is used, Postgres sends a different error message
if replaceAllowed {
if relationType == "view" {
return nil
} else {
return fmt.Errorf(`"%s" is not a view`, newViewName)
}
}
return fmt.Errorf(`relation "%s" already exists`, newViewName)
}
// ValidateNewTableName implements the sql.SchemaObjectNameValidator interface
func (d *PgDatabase) ValidateNewTableName(ctx *sql.Context, newTableName string, skipIfExists bool) (nameAlreadyUsed bool, err error) {
relationExists, _, err := d.doesRelationExist(ctx, newTableName)
if err != nil {
return false, err
}
if !relationExists {
return false, nil
}
if skipIfExists {
return true, nil
}
return true, fmt.Errorf(`relation "%s" already exists`, newTableName)
}
// GenerateIndexName implements the sql.IndexNameGenerator interface with PostgreSQL-compatible naming conventions:
// - UNIQUE indexes: <table>_<col1>[_col2...]_key
// - All other indexes: <table>_<col1>[_col2...]_idx
//
// Collisions are resolved by appending a numeric suffix (1, 2, …) to the base name.
// The collision check uses doesRelationExist so that any schema-level relation — table,
// view, sequence, or another index — blocks a candidate name, matching PostgreSQL's
// behavior where all relations in a schema share one namespace.
func (d *PgDatabase) GenerateIndexName(ctx *sql.Context, tableName string, idxDef sql.IndexDef, _ sql.Table) (string, error) {
colPart := strings.Join(idxDef.ColumnNames(), "_")
suffix := "_idx"
if idxDef.IsUnique() {
suffix = "_key"
}
base := tableName + "_" + colPart + suffix
exists, _, err := d.doesRelationExist(ctx, base)
if err != nil {
return "", err
}
if !exists {
return base, nil
}
for i := 1; ; i++ {
candidate := fmt.Sprintf("%s%d", base, i)
exists, _, err = d.doesRelationExist(ctx, candidate)
if err != nil {
return "", err
}
if !exists {
return candidate, nil
}
}
}
// doesRelationExist tests if a relation with the specified |name| exists in this database. If any relation with that
// name exists, this function returns true for |exists|, the relation type (e.g. index, view, table, sequence) for
// |relationType|. If any problems are encountered looking up a relation, an error is returned in |err|.
func (d *PgDatabase) doesRelationExist(ctx *sql.Context, name string) (exists bool, relationType string, err error) {
lowerName := strings.ToLower(name)
// Resolve the effective schema: when the database was obtained without a schema qualifier
// (e.g. from GMS's plan builder for CREATE INDEX), schemaName is "" and we must fall back to
// the session's current schema so that sequence/view checks use the right namespace.
schema := d.Database.Schema()
if schema == "" {
var err error
schema, err = core.GetCurrentSchema(ctx)
if err != nil || schema == "" {
schema = "public"
}
}
// Tables: use GetTableNames which reads the session's working root directly.
tableNames, err := d.Database.GetTableNames(ctx)
if err != nil {
return false, "", err
}
for _, tableName := range tableNames {
if strings.ToLower(tableName) == lowerName {
return true, "table", nil
}
}
// Sequences: use the session-cached collection so uncommitted sequences are visible.
seqCollection, err := core.GetSequencesCollectionFromContext(ctx, d.Database.Name())
if err != nil {
return false, "", err
}
if seqCollection.HasSequence(ctx, id.NewSequence(schema, name)) {
return true, "sequence", nil
}
// Views: sqle.Database implements sql.ViewDatabase, so call AllViews directly.
views, err := d.Database.AllViews(ctx)
if err != nil {
return false, "", err
}
for _, view := range views {
if strings.ToLower(view.Name) == lowerName {
return true, "view", nil
}
}
// Indexes are per-table; reuse the tableNames slice from the table check above.
for _, tableName := range tableNames {
tbl, ok, err := d.Database.GetTableInsensitive(ctx, tableName)
if err != nil {
return false, "", err
}
if !ok {
continue
}
idxTbl, ok := tbl.(sql.IndexAddressableTable)
if !ok {
continue
}
indexes, err := idxTbl.GetIndexes(ctx)
if err != nil {
return false, "", err
}
for _, idx := range indexes {
if strings.ToLower(idx.ID()) == lowerName {
return true, "index", nil
}
}
}
return false, "", nil
}