-
Notifications
You must be signed in to change notification settings - Fork 8
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
If ISShardedSnapshot is true we now: 1) Create copy of all tables that will be transfered 2) Get all partition key ranges from these tables 3) Send these partitions asynchronously to target commit_hash:ce2c966d9d2b06eb5a4338a5cc47b850333cde07
- Loading branch information
1 parent
0be0bf1
commit b0f9613
Showing
15 changed files
with
436 additions
and
48 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,4 +22,7 @@ type YdbStorageParams struct { | |
|
|
||
| RootCAFiles []string | ||
| TLSEnabled bool | ||
|
|
||
| IsSnapshotSharded bool | ||
| CopyFolder string | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| package ydb | ||
|
|
||
| import ( | ||
| "context" | ||
| "path" | ||
| "strings" | ||
|
|
||
| "github.com/doublecloud/transfer/library/go/core/xerrors" | ||
| "github.com/doublecloud/transfer/pkg/abstract" | ||
| "github.com/ydb-platform/ydb-go-sdk/v3/scheme" | ||
| "github.com/ydb-platform/ydb-go-sdk/v3/table" | ||
| "github.com/ydb-platform/ydb-go-sdk/v3/table/options" | ||
| ) | ||
|
|
||
| const defaultCopyFolder = "data-transfer" | ||
|
|
||
| func (s *Storage) modifyTableName(tablePath string) string { | ||
| return strings.ReplaceAll(tablePath, "/", "_") | ||
| } | ||
|
|
||
| func (s *Storage) BeginSnapshot(ctx context.Context) error { | ||
| if !s.config.IsSnapshotSharded { | ||
| return nil | ||
| } | ||
| if s.config.CopyFolder == "" { | ||
| s.config.CopyFolder = defaultCopyFolder | ||
| } | ||
|
|
||
| tables, err := s.listaAllTablesToTransfer(ctx) | ||
| if err != nil { | ||
| return xerrors.Errorf("Failed to list tables that will be transfered: %w", err) | ||
| } | ||
|
|
||
| if err := s.db.Scheme().MakeDirectory(ctx, s.makeTableDir()); err != nil { | ||
| return xerrors.Errorf("failed to create copy directory: %w", err) | ||
| } | ||
|
|
||
| copyItems := make([]options.CopyTablesOption, len(tables)) | ||
| for i, tableName := range tables { | ||
| tablePath := path.Join(s.config.Database, tableName) | ||
| copyPath := s.makeTablePath("", s.modifyTableName(tableName)) | ||
| copyItems[i] = options.CopyTablesItem(tablePath, copyPath, false) | ||
| } | ||
| return s.db.Table().Do(ctx, func(ctx context.Context, session table.Session) (err error) { | ||
| err = session.CopyTables(ctx, copyItems...) | ||
| if err != nil { | ||
| return xerrors.Errorf("failed to copy tables to transfer directory: %w", err) | ||
| } | ||
| return nil | ||
| }) | ||
| } | ||
|
|
||
| func (s *Storage) EndSnapshot(ctx context.Context) error { | ||
| if !s.config.IsSnapshotSharded { | ||
| return nil | ||
| } | ||
|
|
||
| copyDir := s.makeTableDir() | ||
| content, err := s.db.Scheme().ListDirectory(ctx, copyDir) | ||
| if err != nil { | ||
| return xerrors.Errorf("failed to list copy directory: %w", err) | ||
| } | ||
|
|
||
| err = s.db.Table().Do(ctx, func(ctx context.Context, session table.Session) (err error) { | ||
| for _, copyTable := range content.Children { | ||
| copyPath := s.makeTablePath("", copyTable.Name) | ||
| if copyTable.Type != scheme.EntryTable && copyTable.Type != scheme.EntryColumnTable { | ||
| return xerrors.Errorf("only tables must be present in copy directory, found %v", copyPath) | ||
| } | ||
| if err = session.DropTable(ctx, copyPath); err != nil { | ||
| return xerrors.Errorf("failed to drop copied table %v from transfer directory: %w", copyPath, err) | ||
| } | ||
| } | ||
| return nil | ||
| }) | ||
| if err != nil { | ||
| return xerrors.Errorf("failed to drop copied tables: %w", err) | ||
| } | ||
|
|
||
| if err = s.db.Scheme().RemoveDirectory(ctx, copyDir); err != nil { | ||
| return xerrors.Errorf("failed to remove copy directory: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (s *Storage) ShardTable(ctx context.Context, tableDesc abstract.TableDescription) ([]abstract.TableDescription, error) { | ||
| if !s.config.IsSnapshotSharded { | ||
| return []abstract.TableDescription{tableDesc}, nil | ||
| } | ||
|
|
||
| copyPath := s.makeTablePath(tableDesc.Schema, tableDesc.Name) | ||
| var result []abstract.TableDescription | ||
| err := s.db.Table().Do(ctx, func(ctx context.Context, session table.Session) (err error) { | ||
| tableDescription, err := session.DescribeTable(ctx, copyPath, options.WithShardKeyBounds()) | ||
| if err != nil { | ||
| return xerrors.Errorf("unable to describe table: %w", err) | ||
| } | ||
|
|
||
| result = make([]abstract.TableDescription, len(tableDescription.KeyRanges)) | ||
| for i := range tableDescription.KeyRanges { | ||
| result[i] = tableDesc | ||
| result[i].Offset = uint64(i) | ||
| } | ||
| return nil | ||
| }) | ||
|
|
||
| if err != nil { | ||
| return nil, xerrors.Errorf("unable to schard table %v : %w", copyPath, err) | ||
| } | ||
|
|
||
| return result, nil | ||
| } | ||
|
|
||
| func (s *Storage) makeTablePath(schema, name string) string { | ||
| tableDir := s.makeTableDir() | ||
| if !s.config.IsSnapshotSharded { | ||
| return path.Join(tableDir, schema, name) | ||
| } | ||
| return path.Join(tableDir, schema, s.modifyTableName(name)) | ||
| } | ||
|
|
||
| func (s *Storage) makeTableDir() string { | ||
| if !s.config.IsSnapshotSharded { | ||
| return s.config.Database | ||
| } | ||
| return path.Join(s.config.Database, s.config.CopyFolder) | ||
| } |
Oops, something went wrong.