Merge pull request #540 from ydb-platform/is-table-exists

* Fixed `sugar.IsTableExists` with recursive check directory exists

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,6 @@
+* Fixed `sugar.IsTableExists` with recursive check directory exists
+* Added `sugar.IsDirectoryExists`
+
 ## v3.42.6
 * Implemented `driver.RowsColumnTypeDatabaseTypeName` interface in `internal/xsql.rows` struct
 * Extended `internal/xsql.conn` struct with methods for getting `YDB` metadata

connection.go

@@ -250,6 +250,7 @@ func (c *connection) Scheme() scheme.Client {
 				append(
 					// prepend common params from root config
 					[]schemeConfig.Option{
+						schemeConfig.WithDatabaseName(c.Name()),
 						schemeConfig.With(c.config.Common),
 					},
 					c.schemeOptions...,

internal/scheme/client.go

@@ -19,13 +19,21 @@ import (
 //nolint:nolintlint
 var (
 	errNilClient = xerrors.Wrap(errors.New("scheme client is not initialized"))
+
+	_ interface {
+		Database() string
+	} = (*Client)(nil)
 )
 
 type Client struct {
 	config  config.Config
 	service Ydb_Scheme_V1.SchemeServiceClient
 }
 
+func (c *Client) Database() string {
+	return c.config.Database()
+}
+
 func (c *Client) Close(_ context.Context) error {
 	if c == nil {
 		return xerrors.WithStackTrace(errNilClient)

internal/scheme/config/config.go

@@ -11,14 +11,20 @@ import (
 type Config struct {
 	config.Common
 
-	trace trace.Scheme
+	databaseName string
+	trace        trace.Scheme
 }
 
 // Trace returns trace over scheme client calls
 func (c Config) Trace() trace.Scheme {
 	return c.trace
 }
 
+// Database returns database name
+func (c Config) Database() string {
+	return c.databaseName
+}
+
 type Option func(c *Config)
 
 // WithTrace appends scheme trace to early defined traces
@@ -28,6 +34,13 @@ func WithTrace(trace trace.Scheme, opts ...trace.SchemeComposeOption) Option {
 	}
 }
 
+// WithDatabaseName applies database name
+func WithDatabaseName(dbName string) Option {
+	return func(c *Config) {
+		c.databaseName = dbName
+	}
+}
+
 // With applies common configuration params
 func With(config config.Common) Option {
 	return func(c *Config) {

internal/scheme/helpers/check_exists.go

@@ -0,0 +1,90 @@
+package helpers
+
+import (
+	"context"
+	"fmt"
+	"path"
+	"strings"
+
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
+	"github.com/ydb-platform/ydb-go-sdk/v3/scheme"
+)
+
+type schemeClient interface {
+	ListDirectory(ctx context.Context, path string) (d scheme.Directory, err error)
+}
+
+func databaseName(c schemeClient) string {
+	if name, has := c.(interface {
+		Database() string
+	}); has {
+		return name.Database()
+	}
+	return "/"
+}
+
+func IsDirectoryExists(ctx context.Context, c schemeClient, directory string) (exists bool, _ error) {
+	parentDirectory, childDirectory := path.Split(directory)
+	parentDirectory = strings.TrimRight(parentDirectory, "/")
+	if parentDirectory != databaseName(c) {
+		if exists, err := IsDirectoryExists(ctx, c, parentDirectory); err != nil {
+			return false, xerrors.WithStackTrace(err)
+		} else if !exists {
+			return false, nil
+		}
+	}
+	d, err := c.ListDirectory(ctx, parentDirectory)
+	if err != nil {
+		return false, xerrors.WithStackTrace(err)
+	}
+	for _, e := range d.Children {
+		if e.Name != childDirectory {
+			continue
+		}
+		if e.Type != scheme.EntryDirectory {
+			return false, xerrors.WithStackTrace(fmt.Errorf(
+				"entry '%s' in path '%s' is not a direectory: %s",
+				childDirectory, parentDirectory, e.Type.String(),
+			))
+		}
+		return true, nil
+	}
+	return false, nil
+}
+
+func IsTableExists(ctx context.Context, c schemeClient, absTablePath string) (exists bool, _ error) {
+	if !strings.HasPrefix(absTablePath, databaseName(c)) {
+		return false, xerrors.WithStackTrace(fmt.Errorf(
+			"table path '%s' must be inside database '%s'",
+			absTablePath, databaseName(c),
+		))
+	} else if absTablePath == databaseName(c) {
+		return false, xerrors.WithStackTrace(fmt.Errorf(
+			"table path '%s' cannot be equals database name '%s'",
+			absTablePath, databaseName(c),
+		))
+	}
+	directory, tableName := path.Split(absTablePath)
+	if exists, err := IsDirectoryExists(ctx, c, strings.TrimRight(directory, "/")); err != nil {
+		return false, xerrors.WithStackTrace(err)
+	} else if !exists {
+		return false, nil
+	}
+	d, err := c.ListDirectory(ctx, directory)
+	if err != nil {
+		return false, err
+	}
+	for _, e := range d.Children {
+		if e.Name != tableName {
+			continue
+		}
+		if e.Type != scheme.EntryTable {
+			return false, fmt.Errorf(
+				"entry '%s' in path '%s' is not a table: %s",
+				tableName, directory, e.Type.String(),
+			)
+		}
+		return true, nil
+	}
+	return false, nil
+}

internal/scheme/helpers/check_exists_test.go

@@ -0,0 +1,203 @@
+package helpers
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ydb-platform/ydb-go-sdk/v3/scheme"
+)
+
+type isDirectoryExistsSchemeClient struct {
+	dbName       string
+	existingPath string
+}
+
+func (c isDirectoryExistsSchemeClient) Database() string {
+	return c.dbName
+}
+
+func (c isDirectoryExistsSchemeClient) ListDirectory(ctx context.Context, path string) (
+	d scheme.Directory, err error,
+) {
+	if c.existingPath == path {
+		return scheme.Directory{
+			Entry: scheme.Entry{
+				Name: path,
+				Type: scheme.EntryDirectory,
+			},
+		}, nil
+	}
+	if strings.HasPrefix(c.existingPath, path) {
+		children := strings.Split(strings.TrimLeft(c.existingPath, path), "/")
+		return scheme.Directory{
+			Entry: scheme.Entry{
+				Name: path,
+				Type: scheme.EntryDirectory,
+			},
+			Children: []scheme.Entry{
+				{
+					Name: children[0],
+					Type: scheme.EntryDirectory,
+				},
+			},
+		}, nil
+	}
+	return d, fmt.Errorf("path '%s' not found in '%s'", path, c)
+}
+
+func TestIsDirectoryExists(t *testing.T) {
+	for _, tt := range []struct {
+		checkPath string
+		client    isDirectoryExistsSchemeClient
+		exists    bool
+		err       bool
+	}{
+		{
+			checkPath: "/a/b/c/d",
+			client:    isDirectoryExistsSchemeClient{"/a", "/a/"},
+			exists:    false,
+			err:       false,
+		},
+		{
+			checkPath: "/a/b/c/d",
+			client:    isDirectoryExistsSchemeClient{"/a", "/a/b/"},
+			exists:    false,
+			err:       false,
+		},
+		{
+			checkPath: "/a/b/c/d",
+			client:    isDirectoryExistsSchemeClient{"/a", "/a/b/c/"},
+			exists:    false,
+			err:       false,
+		},
+		{
+			checkPath: "/a/b/c/d",
+			client:    isDirectoryExistsSchemeClient{"/a", "/a/b/c/d/"},
+			exists:    true,
+			err:       false,
+		},
+		{
+			checkPath: "/a/b/c/d",
+			client:    isDirectoryExistsSchemeClient{"/a", "/a/b/c/d/e/"},
+			exists:    true,
+			err:       false,
+		},
+	} {
+		t.Run("", func(t *testing.T) {
+			exists, err := IsDirectoryExists(context.Background(), tt.client, tt.checkPath)
+			if tt.err {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, tt.exists, exists)
+		})
+	}
+}
+
+type isTableExistsSchemeClient struct {
+	dbName    string
+	tablePath string
+}
+
+func (c isTableExistsSchemeClient) Database() string {
+	return c.dbName
+}
+
+func (c isTableExistsSchemeClient) ListDirectory(ctx context.Context, path string) (
+	d scheme.Directory, err error,
+) {
+	if strings.HasPrefix(c.tablePath, path) {
+		children := strings.Split(strings.TrimLeft(c.tablePath, path), "/")
+		switch {
+		case len(children) == 1:
+			return scheme.Directory{
+				Entry: scheme.Entry{
+					Name: path,
+					Type: scheme.EntryDirectory,
+				},
+				Children: []scheme.Entry{
+					{
+						Name: children[0],
+						Type: scheme.EntryTable,
+					},
+				},
+			}, nil
+		case len(children) > 1:
+			return scheme.Directory{
+				Entry: scheme.Entry{
+					Name: path,
+					Type: scheme.EntryDirectory,
+				},
+				Children: []scheme.Entry{
+					{
+						Name: children[0],
+						Type: scheme.EntryDirectory,
+					},
+				},
+			}, nil
+		default:
+			return scheme.Directory{
+				Entry: scheme.Entry{
+					Name: "",
+					Type: scheme.EntryDirectory,
+				},
+			}, nil
+		}
+	}
+	return d, fmt.Errorf("path '%s' not found in '%s'", path, c)
+}
+
+func TestIsTableExists(t *testing.T) {
+	for _, tt := range []struct {
+		checkPath string
+		client    isTableExistsSchemeClient
+		exists    bool
+		err       bool
+	}{
+		{
+			checkPath: "/a/b/c/d",
+			client:    isTableExistsSchemeClient{"/b", "/a/b"},
+			exists:    false,
+			err:       true,
+		},
+		{
+			checkPath: "/a/b/c/d",
+			client:    isTableExistsSchemeClient{"/a", "/a/b"},
+			exists:    false,
+			err:       true,
+		},
+		{
+			checkPath: "/a/b/c/d",
+			client:    isTableExistsSchemeClient{"/a", "/a/b/c"},
+			exists:    false,
+			err:       true,
+		},
+		{
+			checkPath: "/a/b/c/d",
+			client:    isTableExistsSchemeClient{"/a", "/a/b/c/d"},
+			exists:    true,
+			err:       false,
+		},
+		{
+			checkPath: "/a/b/c/d",
+			client:    isTableExistsSchemeClient{"/a", "/a/b/c/d/e"},
+			exists:    false,
+			err:       true,
+		},
+	} {
+		t.Run("", func(t *testing.T) {
+			exists, err := IsTableExists(context.Background(), tt.client, tt.checkPath)
+			if tt.err {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, tt.exists, exists)
+		})
+	}
+}