Improve the way of storing data about collections (FerretDB#1650)

Closes FerretDB#1347.
chilagrow · Dec 25, 2022 · 6fd4368 · 6fd4368
1 parent dd2e78c
commit 6fd4368
Show file tree

Hide file tree

Showing 23 changed files with 826 additions and 570 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,14 +4,22 @@
 
 ### What's Changed
 
-In this release, we made a big change in the way FerretDB stores data in PostgreSQL.
+In this release, we made two big changes in the way FerretDB stores data in PostgreSQL.
+
+The first change is about the format we use to store documents.
 
 Previously, we were storing information about data types in the fields themselves.
 Starting from this release, we store information about data types (document's schema) in a special field.
 
 This will allow us to implement more query push downs in the future.
 
-There are no changes in the API, but since the data is stored in a different way, this change is not backward compatible.
+The second change is about the way FerretDB stores metadata about the collections.
+
+Starting from this release, we store metadata about collections in separate rows of the settings table.
+
+This will allow us to have fewer locks on the settings table and improve performance.
+
+There are no changes in the API, but since the data is stored in a different way, **this change is not backward compatible**.
 Please make a dump of your database before upgrading, delete the databases, upgrade FerretDB, and restore the dump afterwards.
 
 

diff --git a/integration/create_compat_test.go b/integration/create_compat_test.go
@@ -0,0 +1,87 @@
+// Copyright 2021 FerretDB 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 integration
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo/options"
+
+	"github.com/FerretDB/FerretDB/integration/setup"
+	"github.com/FerretDB/FerretDB/integration/shareddata"
+)
+
+// TestCreateCompat tests collection creation compatibility for the cases that are not covered by tests setup.
+func TestCreateCompat(t *testing.T) {
+	t.Helper()
+
+	s := setup.SetupCompatWithOpts(t, &setup.SetupCompatOpts{
+		Providers:                []shareddata.Provider{}, // collections are not needed for this test
+		AddNonExistentCollection: true,
+	})
+
+	// We expect to have only one (non-existent) collection as the result of setup.
+	require.Len(t, s.TargetCollections, 1)
+	require.Len(t, s.CompatCollections, 1)
+
+	targetDB := s.TargetCollections[0].Database()
+	compatDB := s.CompatCollections[0].Database()
+
+	// Test collection creation in non-existent database.
+	err := targetDB.Drop(s.Ctx)
+	require.NoError(t, err)
+
+	err = compatDB.Drop(s.Ctx)
+	require.NoError(t, err)
+
+	collName := "in-non-existent-db"
+
+	// schema in case of Tigris.
+	schema := fmt.Sprintf(`{
+				"title": "%s",
+				"description": "Create Collection In Non-Existent Database",
+				"primary_key": ["_id"],
+				"properties": {
+					"_id": {"type": "string"}
+				}
+			}`, collName,
+	)
+	opts := options.CreateCollectionOptions{
+		Validator: bson.D{{"$tigrisSchemaString", schema}},
+	}
+
+	targetErr := targetDB.CreateCollection(s.Ctx, collName, &opts)
+	if targetErr != nil {
+		if errorTextContains(targetErr,
+			`support for field "validator" is not implemented yet`,
+		) {
+			targetErr = targetDB.CreateCollection(s.Ctx, collName)
+		}
+
+		require.NoError(t, targetErr)
+	}
+
+	compatErr := compatDB.CreateCollection(s.Ctx, collName)
+	require.Equal(t, targetErr, compatErr)
+
+	targetNames, targetErr := targetDB.ListCollectionNames(s.Ctx, bson.D{})
+	compatNames, compatErr := compatDB.ListCollectionNames(s.Ctx, bson.D{})
+	require.NoError(t, compatErr)
+	require.Equal(t, targetErr, compatErr)
+	require.Equal(t, targetNames, compatNames)
+}
diff --git a/integration/update_compat_test.go b/integration/update_compat_test.go
@@ -306,7 +306,7 @@ type updateCurrentDateCompatTestCase struct {
 func testUpdateCurrentDateCompat(t *testing.T, testCases map[string]updateCurrentDateCompatTestCase) {
 	t.Helper()
 
-	maxDifference := 4 * time.Minute
+	maxDifference := 2 * time.Minute
 
 	for name, tc := range testCases {
 		name, tc := name, tc

diff --git a/internal/handlers/pg/msg_create.go b/internal/handlers/pg/msg_create.go
@@ -75,29 +75,27 @@ func (h *Handler) MsgCreate(ctx context.Context, msg *wire.OpMsg) (*wire.OpMsg,
 	}
 
 	err = h.PgPool.InTransactionRetry(ctx, func(tx pgx.Tx) error {
-		if err := pgdb.CreateDatabaseIfNotExists(ctx, tx, db); err != nil {
-			switch {
-			case errors.Is(pgdb.ErrInvalidDatabaseName, err):
-				msg := fmt.Sprintf("Invalid namespace: %s.%s", db, collection)
-				return common.NewCommandErrorMsg(common.ErrInvalidNamespace, msg)
-			default:
-				return lazyerrors.Error(err)
-			}
-		}
+		err = pgdb.CreateCollection(ctx, tx, db, collection)
+
+		switch {
+		case err == nil:
+			return nil
+
+		case errors.Is(err, pgdb.ErrAlreadyExist):
+			msg := fmt.Sprintf("Collection %s.%s already exists.", db, collection)
+			return common.NewCommandErrorMsg(common.ErrNamespaceExists, msg)
+
+		case errors.Is(err, pgdb.ErrInvalidDatabaseName):
+			msg := fmt.Sprintf("Invalid namespace: %s.%s", db, collection)
+			return common.NewCommandErrorMsg(common.ErrInvalidNamespace, msg)
+
+		case errors.Is(err, pgdb.ErrInvalidCollectionName):
+			msg := fmt.Sprintf("Invalid collection name: '%s.%s'", db, collection)
+			return common.NewCommandErrorMsg(common.ErrInvalidNamespace, msg)
 
-		if err := pgdb.CreateCollection(ctx, tx, db, collection); err != nil {
-			switch {
-			case errors.Is(err, pgdb.ErrAlreadyExist):
-				msg := fmt.Sprintf("Collection %s.%s already exists.", db, collection)
-				return common.NewCommandErrorMsg(common.ErrNamespaceExists, msg)
-			case errors.Is(err, pgdb.ErrInvalidTableName):
-				msg := fmt.Sprintf("Invalid collection name: '%s.%s'", db, collection)
-				return common.NewCommandErrorMsg(common.ErrInvalidNamespace, msg)
-			default:
-				return lazyerrors.Error(err)
-			}
+		default:
+			return lazyerrors.Error(err)
 		}
-		return nil
 	})
 	if err != nil {
 		return nil, err

diff --git a/internal/handlers/pg/msg_delete.go b/internal/handlers/pg/msg_delete.go
@@ -183,7 +183,7 @@ func (h *Handler) execDelete(ctx context.Context, sp *pgdb.SQLParam, filter *typ
 	err := h.PgPool.InTransaction(ctx, func(tx pgx.Tx) error {
 		var it iterator.Interface[uint32, *types.Document]
 		var err error
-		it, err = h.PgPool.GetDocuments(ctx, tx, sp)
+		it, err = pgdb.GetDocuments(ctx, tx, sp)
 		if err != nil {
 			return err
 		}

diff --git a/internal/handlers/pg/msg_find.go b/internal/handlers/pg/msg_find.go
@@ -116,7 +116,7 @@ func (h *Handler) fetchAndFilterDocs(ctx context.Context, tx pgx.Tx, sqlParam *p
 
 	var it iterator.Interface[uint32, *types.Document]
 
-	it, err := h.PgPool.GetDocuments(ctx, tx, sqlParam)
+	it, err := pgdb.GetDocuments(ctx, tx, sqlParam)
 	if err != nil {
 		return nil, err
 	}

diff --git a/internal/handlers/pg/msg_insert.go b/internal/handlers/pg/msg_insert.go
@@ -115,7 +115,7 @@ func (h *Handler) insert(ctx context.Context, sp *pgdb.SQLParam, doc any) error
 		return nil
 	}
 
-	if errors.Is(pgdb.ErrInvalidTableName, err) || errors.Is(pgdb.ErrInvalidDatabaseName, err) {
+	if errors.Is(err, pgdb.ErrInvalidCollectionName) || errors.Is(err, pgdb.ErrInvalidDatabaseName) {
 		msg := fmt.Sprintf("Invalid namespace: %s.%s", sp.DB, sp.Collection)
 		return common.NewCommandErrorMsg(common.ErrInvalidNamespace, msg)
 	}

diff --git a/internal/handlers/pg/msg_update.go b/internal/handlers/pg/msg_update.go
@@ -20,7 +20,6 @@ import (
 	"fmt"
 
 	"github.com/jackc/pgx/v4"
-	"go.uber.org/zap"
 
 	"github.com/FerretDB/FerretDB/internal/handlers/common"
 	"github.com/FerretDB/FerretDB/internal/handlers/pg/pgdb"
@@ -67,18 +66,17 @@ func (h *Handler) MsgUpdate(ctx context.Context, msg *wire.OpMsg) (*wire.OpMsg,
 		return nil, err
 	}
 
-	created, err := pgdb.CreateCollectionIfNotExist(ctx, h.PgPool, sp.DB, sp.Collection)
+	err = h.PgPool.InTransactionRetry(ctx, func(tx pgx.Tx) error {
+		return pgdb.CreateCollectionIfNotExists(ctx, tx, sp.DB, sp.Collection)
+	})
 	if err != nil {
-		if errors.Is(pgdb.ErrInvalidTableName, err) ||
-			errors.Is(pgdb.ErrInvalidDatabaseName, err) {
+		if errors.Is(err, pgdb.ErrInvalidCollectionName) ||
+			errors.Is(err, pgdb.ErrInvalidDatabaseName) {
 			msg := fmt.Sprintf("Invalid namespace: %s.%s", sp.DB, sp.Collection)
 			return nil, common.NewCommandErrorMsg(common.ErrInvalidNamespace, msg)
 		}
 		return nil, err
 	}
-	if created {
-		h.L.Info("Created table.", zap.String("schema", sp.DB), zap.String("table", sp.Collection))
-	}
 
 	var matched, modified int32
 	var upserted types.Array