diff --git a/pkg/ccl/backupccl/targets.go b/pkg/ccl/backupccl/targets.go index 7982096de809..134981658b81 100644 --- a/pkg/ccl/backupccl/targets.go +++ b/pkg/ccl/backupccl/targets.go @@ -444,13 +444,16 @@ func selectTargets( return matched.Descs, matched.RequestedDBs, nil, nil } +// EntryFiles is a group of sst files of a backup table range +type EntryFiles []roachpb.ImportRequest_File + // BackupTableEntry wraps information of a table retrieved // from backup manifests. // exported to cliccl for exporting data directly from backup sst. type BackupTableEntry struct { Desc catalog.TableDescriptor Span roachpb.Span - Files []roachpb.ImportRequest_File + Files []EntryFiles } // MakeBackupTableEntry looks up the descriptor of fullyQualifiedTableName @@ -525,11 +528,15 @@ func MakeBackupTableEntry( return BackupTableEntry{}, errors.Wrapf(err, "making spans for table %s", fullyQualifiedTableName) } - res := BackupTableEntry{ + backupTableEntry := BackupTableEntry{ tbDesc, tablePrimaryIndexSpan, - entry[0].Files, + make([]EntryFiles, 0), } - return res, nil + for _, e := range entry { + backupTableEntry.Files = append(backupTableEntry.Files, e.Files) + } + + return backupTableEntry, nil } diff --git a/pkg/ccl/cliccl/debug_backup.go b/pkg/ccl/cliccl/debug_backup.go index ebf504611841..6941550a9f22 100644 --- a/pkg/ccl/cliccl/debug_backup.go +++ b/pkg/ccl/cliccl/debug_backup.go @@ -52,12 +52,16 @@ import ( "github.com/spf13/cobra" ) -var externalIODir string -var exportTableName string -var readTime string -var destination string -var format string -var nullas string +// debugBackupArgs captures the parameters of the `debug backup` command. +var debugBackupArgs struct { + externalIODir string + + exportTableName string + readTime string + destination string + format string + nullas string +} func init() { @@ -105,42 +109,42 @@ func init() { backupFlags := backupCmds.Flags() backupFlags.StringVarP( - &externalIODir, + &debugBackupArgs.externalIODir, cliflags.ExternalIODir.Name, cliflags.ExternalIODir.Shorthand, "", /*value*/ cliflags.ExternalIODir.Usage()) exportDataCmd.Flags().StringVarP( - &exportTableName, + &debugBackupArgs.exportTableName, cliflags.ExportTableTarget.Name, cliflags.ExportTableTarget.Shorthand, "", /*value*/ cliflags.ExportTableTarget.Usage()) exportDataCmd.Flags().StringVarP( - &readTime, + &debugBackupArgs.readTime, cliflags.ReadTime.Name, cliflags.ReadTime.Shorthand, "", /*value*/ cliflags.ReadTime.Usage()) exportDataCmd.Flags().StringVarP( - &destination, + &debugBackupArgs.destination, cliflags.ExportDestination.Name, cliflags.ExportDestination.Shorthand, "", /*value*/ cliflags.ExportDestination.Usage()) exportDataCmd.Flags().StringVarP( - &format, + &debugBackupArgs.format, cliflags.ExportTableFormat.Name, cliflags.ExportTableFormat.Shorthand, "csv", /*value*/ cliflags.ExportTableFormat.Usage()) exportDataCmd.Flags().StringVarP( - &nullas, + &debugBackupArgs.nullas, cliflags.ExportCSVNullas.Name, cliflags.ExportCSVNullas.Shorthand, "null", /*value*/ @@ -165,10 +169,10 @@ func newBlobFactory(ctx context.Context, dialing roachpb.NodeID) (blobs.BlobClie if dialing != 0 { return nil, errors.Errorf("accessing node %d during nodelocal access is unsupported for CLI inspection; only local access is supported with nodelocal://self", dialing) } - if externalIODir == "" { - externalIODir = filepath.Join(server.DefaultStorePath, "extern") + if debugBackupArgs.externalIODir == "" { + debugBackupArgs.externalIODir = filepath.Join(server.DefaultStorePath, "extern") } - return blobs.NewLocalClient(externalIODir) + return blobs.NewLocalClient(debugBackupArgs.externalIODir) } func externalStorageFromURIFactory( @@ -302,10 +306,10 @@ func runListIncrementalCmd(cmd *cobra.Command, args []string) error { func runExportDataCmd(cmd *cobra.Command, args []string) error { - if exportTableName == "" { + if debugBackupArgs.exportTableName == "" { return errors.New("export data requires table name specified by --table flag") } - fullyQualifiedTableName := strings.ToLower(exportTableName) + fullyQualifiedTableName := strings.ToLower(debugBackupArgs.exportTableName) manifestPaths := args ctx := context.Background() @@ -318,9 +322,9 @@ func runExportDataCmd(cmd *cobra.Command, args []string) error { manifests = append(manifests, manifest) } - endTime, err := evalAsOfTimestamp(readTime) + endTime, err := evalAsOfTimestamp(debugBackupArgs.readTime) if err != nil { - return errors.Wrapf(err, "eval as of timestamp %s", readTime) + return errors.Wrapf(err, "eval as of timestamp %s", debugBackupArgs.readTime) } codec := keys.TODOSQLCodec @@ -364,74 +368,36 @@ func evalAsOfTimestamp(readTime string) (hlc.Timestamp, error) { func showData( ctx context.Context, entry backupccl.BackupTableEntry, endTime hlc.Timestamp, codec keys.SQLCodec, -) (err error) { - - iters, cleanup, err := makeIters(ctx, entry) - if err != nil { - return errors.Wrapf(err, "make iters") - } - defer func() { - cleanupErr := cleanup() - if err == nil { - err = cleanupErr - } - }() - - iter := storage.MakeMultiIterator(iters) - defer iter.Close() - - rf, err := makeRowFetcher(ctx, entry, codec) - if err != nil { - return errors.Wrapf(err, "make row fetcher") - } - defer rf.Close(ctx) - - startKeyMVCC, endKeyMVCC := storage.MVCCKey{Key: entry.Span.Key}, storage.MVCCKey{Key: entry.Span.EndKey} - kvFetcher := row.MakeBackupSSTKVFetcher(startKeyMVCC, endKeyMVCC, iter, endTime) - - if err := rf.StartScanFrom(ctx, &kvFetcher); err != nil { - return errors.Wrapf(err, "row fetcher starts scan") - } +) error { + buf := bytes.NewBuffer([]byte{}) var writer *csv.Writer - if format != "csv" { + if debugBackupArgs.format != "csv" { return errors.Newf("only exporting to csv format is supported") } - - buf := bytes.NewBuffer([]byte{}) - if destination == "" { + if debugBackupArgs.destination == "" { writer = csv.NewWriter(os.Stdout) } else { writer = csv.NewWriter(buf) } - for { - datums, _, _, err := rf.NextRowDecoded(ctx) - if err != nil { - return errors.Wrapf(err, "decode row") - } - if datums == nil { - break - } - row := make([]string, datums.Len()) - for i, datum := range datums { - if datum == tree.DNull { - row[i] = nullas - } else { - row[i] = datum.String() - } - } - if err := writer.Write(row); err != nil { + rf, err := makeRowFetcher(ctx, entry, codec) + if err != nil { + return errors.Wrapf(err, "make row fetcher") + } + defer rf.Close(ctx) + + for _, files := range entry.Files { + if err := processEntryFiles(ctx, rf, files, entry.Span, endTime, writer); err != nil { return err } - writer.Flush() } - if destination != "" { - dir, file := filepath.Split(destination) + if debugBackupArgs.destination != "" { + dir, file := filepath.Split(debugBackupArgs.destination) store, err := externalStorageFromURIFactory(ctx, dir, security.RootUserName()) if err != nil { - return errors.Wrapf(err, "unable to open store to write files: %s", destination) + return errors.Wrapf(err, "unable to open store to write files: %s", debugBackupArgs.destination) } if err = store.WriteFile(ctx, file, bytes.NewReader(buf.Bytes())); err != nil { _ = store.Close() @@ -439,15 +405,15 @@ func showData( } return store.Close() } - return err + return nil } func makeIters( - ctx context.Context, entry backupccl.BackupTableEntry, + ctx context.Context, files backupccl.EntryFiles, ) ([]storage.SimpleMVCCIterator, func() error, error) { - iters := make([]storage.SimpleMVCCIterator, len(entry.Files)) - dirStorage := make([]cloud.ExternalStorage, len(entry.Files)) - for i, file := range entry.Files { + iters := make([]storage.SimpleMVCCIterator, len(files)) + dirStorage := make([]cloud.ExternalStorage, len(files)) + for i, file := range files { var err error clusterSettings := cluster.MakeClusterSettings() dirStorage[i], err = cloudimpl.MakeExternalStorage(ctx, file.Dir, base.ExternalIODirConfig{}, @@ -512,6 +478,59 @@ func makeRowFetcher( return rf, nil } +func processEntryFiles( + ctx context.Context, + rf row.Fetcher, + files backupccl.EntryFiles, + span roachpb.Span, + endTime hlc.Timestamp, + writer *csv.Writer, +) (err error) { + + iters, cleanup, err := makeIters(ctx, files) + defer func() { + if cleanupErr := cleanup(); err == nil { + err = cleanupErr + } + }() + if err != nil { + return errors.Wrapf(err, "make iters") + } + + iter := storage.MakeMultiIterator(iters) + defer iter.Close() + + startKeyMVCC, endKeyMVCC := storage.MVCCKey{Key: span.Key}, storage.MVCCKey{Key: span.EndKey} + kvFetcher := row.MakeBackupSSTKVFetcher(startKeyMVCC, endKeyMVCC, iter, endTime) + + if err := rf.StartScanFrom(ctx, &kvFetcher); err != nil { + return errors.Wrapf(err, "row fetcher starts scan") + } + + for { + datums, _, _, err := rf.NextRowDecoded(ctx) + if err != nil { + return errors.Wrapf(err, "decode row") + } + if datums == nil { + break + } + rowDisplay := make([]string, datums.Len()) + for i, datum := range datums { + if datum == tree.DNull { + rowDisplay[i] = debugBackupArgs.nullas + } else { + rowDisplay[i] = datum.String() + } + } + if err := writer.Write(rowDisplay); err != nil { + return err + } + writer.Flush() + } + return nil +} + type backupMetaDisplayMsg backupccl.BackupManifest type backupFileDisplayMsg backupccl.BackupManifest_File diff --git a/pkg/ccl/cliccl/debug_backup_test.go b/pkg/ccl/cliccl/debug_backup_test.go index 6bdf13fdefaf..b4e15d030a27 100644 --- a/pkg/ccl/cliccl/debug_backup_test.go +++ b/pkg/ccl/cliccl/debug_backup_test.go @@ -231,7 +231,7 @@ func TestListIncremental(t *testing.T) { checkExpectedOutput(t, buf.String(), out) } -func TestShowData(t *testing.T) { +func TestExportData(t *testing.T) { defer leaktest.AfterTest(t)() defer log.Scope(t).Close(t) @@ -340,7 +340,71 @@ func TestShowData(t *testing.T) { } } -func TestShowDataAOST(t *testing.T) { +func TestExportDataWithMultipleRanges(t *testing.T) { + defer leaktest.AfterTest(t)() + defer log.Scope(t).Close(t) + + c := cli.NewCLITest(cli.TestCLIParams{T: t, NoServer: true}) + defer c.Cleanup() + + ctx := context.Background() + dir, cleanFn := testutils.TempDir(t) + defer cleanFn() + srv, db, _ := serverutils.StartServer(t, base.TestServerArgs{ExternalIODir: dir, Insecure: true}) + defer srv.Stopper().Stop(ctx) + + sqlDB := sqlutils.MakeSQLRunner(db) + sqlDB.Exec(t, `CREATE DATABASE testDB`) + sqlDB.Exec(t, `USE testDB`) + sqlDB.Exec(t, `CREATE TABLE fooTable(id int PRIMARY KEY)`) + sqlDB.Exec(t, `INSERT INTO fooTable select * from generate_series(1,10)`) + sqlDB.Exec(t, `ALTER TABLE fooTable SPLIT AT VALUES (2), (5), (7)`) + + const backupPath = "nodelocal://0/fooFolder" + sqlDB.Exec(t, `BACKUP TABLE fooTable TO $1 `, backupPath) + + var rangeNum int + sqlDB.QueryRow(t, `SELECT count(*) from [SHOW RANGES from TABLE fooTable]`).Scan(&rangeNum) + require.Equal(t, 4, rangeNum) + sqlDB.QueryRow(t, `SELECT count(*) from [SHOW BACKUP FILES $1]`, backupPath).Scan(&rangeNum) + require.Equal(t, 4, rangeNum) + + sqlDB.Exec(t, `ALTER TABLE fooTable ADD COLUMN active BOOL DEFAULT false`) + sqlDB.Exec(t, `INSERT INTO fooTable select * from generate_series(11,15)`) + ts := hlc.Timestamp{WallTime: timeutil.Now().UnixNano()} + sqlDB.Exec(t, fmt.Sprintf(`BACKUP TABLE fooTable TO $1 AS OF SYSTEM TIME '%s'`, ts.AsOfSystemTime()), backupPath) + + sqlDB.QueryRow(t, `SELECT count(*) from [SHOW RANGES from TABLE fooTable]`).Scan(&rangeNum) + require.Equal(t, 4, rangeNum) + sqlDB.QueryRow(t, `SELECT count(*) from [SHOW BACKUP FILES $1]`, backupPath).Scan(&rangeNum) + require.Equal(t, 8, rangeNum) + + t.Run("export-data-with-multiple-ranges", func(t *testing.T) { + out, err := c.RunWithCapture(fmt.Sprintf("debug backup export %s --table=testDB.public.fooTable --external-io-dir=%s", + backupPath, + dir)) + require.NoError(t, err) + var expectedOut string + for i := 1; i <= 10; i++ { + expectedOut = fmt.Sprintf("%s%d\n", expectedOut, i) + } + checkExpectedOutput(t, expectedOut, out) + }) + + t.Run("export-data-with-multiple-ranges-in-incremental-backups", func(t *testing.T) { + out, err := c.RunWithCapture(fmt.Sprintf("debug backup export %s %s --table=testDB.public.fooTable --external-io-dir=%s", + backupPath, backupPath+ts.GoTime().Format(backupccl.DateBasedIncFolderName), + dir)) + require.NoError(t, err) + var expectedOut string + for i := 1; i <= 15; i++ { + expectedOut = fmt.Sprintf("%s%d,false\n", expectedOut, i) + } + checkExpectedOutput(t, expectedOut, out) + }) +} + +func TestExportDataAOST(t *testing.T) { defer leaktest.AfterTest(t)() defer log.Scope(t).Close(t) diff --git a/pkg/cmd/generate-binary/main.go b/pkg/cmd/generate-binary/main.go index 982dd4728a8e..4e9d5b0f172d 100644 --- a/pkg/cmd/generate-binary/main.go +++ b/pkg/cmd/generate-binary/main.go @@ -200,6 +200,11 @@ var inputs = map[string][]string{ "42.0", "420000", "420000.0", + "6000500000000.0000000", + "10000", + "800000000", + "9E+4", + "99E100", }, "'%s'::float8": { diff --git a/pkg/sql/pgwire/encoding_test.go b/pkg/sql/pgwire/encoding_test.go index a750c0b7b76f..e2bd2ae813d7 100644 --- a/pkg/sql/pgwire/encoding_test.go +++ b/pkg/sql/pgwire/encoding_test.go @@ -156,7 +156,7 @@ func TestEncodings(t *testing.T) { } got := verifyLen(t) if !bytes.Equal(got, tc.TextAsBinary) { - t.Errorf("unexpected text encoding:\n\t%q found,\n\t%q expected", got, tc.Text) + t.Errorf("unexpected text encoding:\n\t%q found,\n\t%q expected", got, tc.TextAsBinary) } } }) diff --git a/pkg/sql/pgwire/testdata/encodings.json b/pkg/sql/pgwire/testdata/encodings.json index 708ef742aa9b..e1827a2d978c 100644 --- a/pkg/sql/pgwire/testdata/encodings.json +++ b/pkg/sql/pgwire/testdata/encodings.json @@ -580,6 +580,41 @@ "TextAsBinary": [52, 50, 48, 48, 48, 48, 46, 48], "Binary": [0, 1, 0, 1, 0, 0, 0, 1, 0, 42] }, + { + "SQL": "'6000500000000.0000000'::decimal", + "Oid": 1700, + "Text": "6000500000000.0000000", + "TextAsBinary": [54, 48, 48, 48, 53, 48, 48, 48, 48, 48, 48, 48, 48, 46, 48, 48, 48, 48, 48, 48, 48], + "Binary": [0, 2, 0, 3, 0, 0, 0, 7, 0, 6, 0, 5] + }, + { + "SQL": "'10000'::decimal", + "Oid": 1700, + "Text": "10000", + "TextAsBinary": [49, 48, 48, 48, 48], + "Binary": [0, 1, 0, 1, 0, 0, 0, 0, 0, 1] + }, + { + "SQL": "'800000000'::decimal", + "Oid": 1700, + "Text": "800000000", + "TextAsBinary": [56, 48, 48, 48, 48, 48, 48, 48, 48], + "Binary": [0, 1, 0, 2, 0, 0, 0, 0, 0, 8] + }, + { + "SQL": "'9E+4'::decimal", + "Oid": 1700, + "Text": "9E+4", + "TextAsBinary": [57, 69, 43, 52], + "Binary": [0, 1, 0, 1, 0, 0, 0, 0, 0, 9] + }, + { + "SQL": "'99E100'::decimal", + "Oid": 1700, + "Text": "9.9E+101", + "TextAsBinary": [57, 46, 57, 69, 43, 49, 48, 49], + "Binary": [0, 1, 0, 25, 0, 0, 0, 0, 0, 99] + }, { "SQL": "'{-000.000,-0000021234.23246346000000,-1.2,.0,.1,.1234}'::decimal[]", "Oid": 1231, diff --git a/pkg/sql/pgwire/testdata/pgtest/decimal b/pkg/sql/pgwire/testdata/pgtest/decimal index 2ae73557a020..c6c3894e63d4 100644 --- a/pkg/sql/pgwire/testdata/pgtest/decimal +++ b/pkg/sql/pgwire/testdata/pgtest/decimal @@ -35,3 +35,51 @@ ReadyForQuery {"Type":"DataRow","Values":[{"text":"0"}]} {"Type":"CommandComplete","CommandTag":"SELECT 1"} {"Type":"ReadyForQuery","TxStatus":"I"} + +# [0, 1, 0, 1, 0, 0, 0, 0, 0, 1] = binary 1E+4 +# [0, 1, 0, 1, 0, 0, 0, 0, 0, 10] = binary 10E+4 +send +Parse {"Name": "s1", "Query": "SELECT 10000::decimal, $1::decimal, $2::decimal"} +Bind {"DestinationPortal": "p1", "PreparedStatement": "s1", "ParameterFormatCodes": [1], "Parameters": [[0, 1, 0, 1, 0, 0, 0, 0, 0, 1], [0, 1, 0, 1, 0, 0, 0, 0, 0, 10]]} +Execute {"Portal": "p1"} +Sync +---- + +# CockroachDB intentionally uses exponents for decimals like 1E+4, as +# oppposed to Postgres, which returns 10000. +until crdb_only +ReadyForQuery +---- +{"Type":"ParseComplete"} +{"Type":"BindComplete"} +{"Type":"DataRow","Values":[{"text":"10000"},{"text":"1E+4"},{"text":"1.0E+5"}]} +{"Type":"CommandComplete","CommandTag":"SELECT 1"} +{"Type":"ReadyForQuery","TxStatus":"I"} + +until noncrdb_only +ReadyForQuery +---- +{"Type":"ParseComplete"} +{"Type":"BindComplete"} +{"Type":"DataRow","Values":[{"text":"10000"},{"text":"10000"},{"text":"100000"}]} +{"Type":"CommandComplete","CommandTag":"SELECT 1"} +{"Type":"ReadyForQuery","TxStatus":"I"} + +# ResultFormatCodes [1] = FormatBinary +# [0, 1, 0, 1, 0, 0, 0, 0, 0, 1] = binary 1E+4 +# [0, 1, 0, 1, 0, 0, 0, 0, 0, 10] = binary 10E+4 +send +Parse {"Name": "s2", "Query": "SELECT 10000::decimal, $1::decimal, $2::decimal"} +Bind {"DestinationPortal": "p2", "PreparedStatement": "s2", "ParameterFormatCodes": [1], "Parameters": [[0, 1, 0, 1, 0, 0, 0, 0, 0, 1], [0, 1, 0, 1, 0, 0, 0, 0, 0, 10]], "ResultFormatCodes": [1,1, 1]} +Execute {"Portal": "p2"} +Sync +---- + +until +ReadyForQuery +---- +{"Type":"ParseComplete"} +{"Type":"BindComplete"} +{"Type":"DataRow","Values":[{"binary":"00010001000000000001"},{"binary":"00010001000000000001"},{"binary":"0001000100000000000a"}]} +{"Type":"CommandComplete","CommandTag":"SELECT 1"} +{"Type":"ReadyForQuery","TxStatus":"I"} diff --git a/pkg/sql/pgwire/types.go b/pkg/sql/pgwire/types.go index d46d102edf97..0e46e5a8c074 100644 --- a/pkg/sql/pgwire/types.go +++ b/pkg/sql/pgwire/types.go @@ -353,6 +353,12 @@ func (b *writeBuffer) writeBinaryDatum( return ndigit } + // The dscale is defined as number of digits (in base 10) visible + // after the decimal separator, so it can't be negative. + if alloc.pgNum.Dscale < 0 { + alloc.pgNum.Dscale = 0 + } + b.putInt32(int32(2 * (4 + alloc.pgNum.Ndigits))) b.putInt16(alloc.pgNum.Ndigits) b.putInt16(alloc.pgNum.Weight)