Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

release-25.1: rowexec: add concurrency to ingestIndexEntries #139216

Merged
merged 3 commits into from
Jan 16, 2025
Merged

release-25.1: rowexec: add concurrency to ingestIndexEntries #139216

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8f28d08
sql: remove unnecessary test
rafiss Jan 9, 2025
770f281
rowexec: add concurrency to ingestIndexEntries
rafiss Jan 9, 2025
5662367
roachtest: make indexschemachange test more consistent
rafiss Jan 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions pkg/cmd/roachtest/tests/schemachange.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -303,7 +303,7 @@ func makeIndexAddTpccTest(
Cluster: spec,
CompatibleClouds: registry.AllExceptAWS,
Suites: registry.Suites(registry.Nightly),
Leases: registry.MetamorphicLeases,
Leases: registry.DefaultLeases,
Timeout: length * 3,
Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
runTPCC(ctx, t, t.L(), c, tpccOptions{
Expand All @@ -313,13 +313,15 @@ func makeIndexAddTpccTest(
ExtraRunArgs: fmt.Sprintf("--wait=false --tolerate-errors --workers=%d", warehouses),
During: func(ctx context.Context) error {
return runAndLogStmts(ctx, t, c, "addindex", []string{
`SET CLUSTER SETTING bulkio.index_backfill.ingest_concurrency = 2;`,
`CREATE UNIQUE INDEX ON tpcc.order (o_entry_d, o_w_id, o_d_id, o_carrier_id, o_id);`,
`CREATE INDEX ON tpcc.order (o_carrier_id);`,
`CREATE INDEX ON tpcc.customer (c_last, c_first);`,
})
},
Duration: length,
SetupType: usingImport,
DisableDefaultScheduledBackup: true,
Duration: length,
SetupType: usingImport,
})
},
}
Expand Down
57 changes: 39 additions & 18 deletions pkg/sql/rowexec/indexbackfiller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,6 @@ import (
type indexBackfiller struct {
backfill.IndexBackfiller

adder kvserverbase.BulkAdder

desc catalog.TableDescriptor

spec execinfrapb.BackfillerSpec
Expand All @@ -53,12 +51,29 @@ var _ execinfra.Processor = &indexBackfiller{}

var backfillerBufferSize = settings.RegisterByteSizeSetting(
settings.ApplicationLevel,
"schemachanger.backfiller.buffer_size", "the initial size of the BulkAdder buffer handling index backfills", 32<<20,
"schemachanger.backfiller.buffer_size",
"the initial size of the BulkAdder buffer handling index backfills",
32<<20,
)

var backfillerMaxBufferSize = settings.RegisterByteSizeSetting(
settings.ApplicationLevel,
"schemachanger.backfiller.max_buffer_size", "the maximum size of the BulkAdder buffer handling index backfills", 512<<20,
"schemachanger.backfiller.max_buffer_size",
"the maximum size of the BulkAdder buffer handling index backfills",
512<<20,
)

// indexBackfillIngestConcurrency is the number of goroutines to use for
// the ingestion step of the index backfiller; these are the goroutines
// that write the index entries in bulk. Since that is mostly I/O bound,
// adding concurrency allows some of the computational work to occur in
// parallel.
var indexBackfillIngestConcurrency = settings.RegisterIntSetting(
settings.ApplicationLevel,
"bulkio.index_backfill.ingest_concurrency",
"the number of goroutines to use for bulk adding index entries",
2,
settings.PositiveInt, /* validateFn */
)

func newIndexBackfiller(
Expand Down Expand Up @@ -185,8 +200,7 @@ func (ib *indexBackfiller) ingestIndexEntries(
if err != nil {
return err
}
ib.adder = adder
defer ib.adder.Close(ctx)
defer adder.Close(ctx)

// Synchronizes read and write access on completedSpans which is updated on a
// BulkAdder flush, but is read when progress is being sent back to the
Expand Down Expand Up @@ -241,7 +255,7 @@ func (ib *indexBackfiller) ingestIndexEntries(

for indexBatch := range indexEntryCh {
for _, indexEntry := range indexBatch.indexEntries {
if err := ib.adder.Add(ctx, indexEntry.Key, indexEntry.Value.RawBytes); err != nil {
if err := adder.Add(ctx, indexEntry.Key, indexEntry.Value.RawBytes); err != nil {
return ib.wrapDupError(ctx, err)
}
}
Expand All @@ -261,7 +275,7 @@ func (ib *indexBackfiller) ingestIndexEntries(

knobs := &ib.flowCtx.Cfg.TestingKnobs
if knobs.BulkAdderFlushesEveryBatch {
if err := ib.adder.Flush(ctx); err != nil {
if err := adder.Flush(ctx); err != nil {
return ib.wrapDupError(ctx, err)
}
pushProgress()
Expand All @@ -283,7 +297,7 @@ func (ib *indexBackfiller) ingestIndexEntries(
return err
}

if err := ib.adder.Flush(ctx); err != nil {
if err := adder.Flush(ctx); err != nil {
return ib.wrapDupError(ctx, err)
}

Expand All @@ -296,8 +310,11 @@ func (ib *indexBackfiller) ingestIndexEntries(
func (ib *indexBackfiller) runBackfill(
ctx context.Context, progCh chan execinfrapb.RemoteProducerMetadata_BulkProcessorProgress,
) error {
const indexEntriesChBufferSize = 10
ingestConcurrency := indexBackfillIngestConcurrency.Get(&ib.flowCtx.Cfg.Settings.SV)

// Used to send index entries to the KV layer.
indexEntriesCh := make(chan indexEntryBatch, 10)
indexEntriesCh := make(chan indexEntryBatch, indexEntriesChBufferSize)

// This group holds the go routines that are responsible for producing index
// entries and ingesting the KVs into storage.
Expand All @@ -315,14 +332,18 @@ func (ib *indexBackfiller) runBackfill(
return nil
})

// Ingest the index entries that are emitted to the chan.
group.GoCtx(func(ctx context.Context) error {
err := ib.ingestIndexEntries(ctx, indexEntriesCh, progCh)
if err != nil {
return errors.Wrap(err, "failed to ingest index entries during backfill")
}
return nil
})
// Ingest the index entries that are emitted to the chan. We use multiple
// goroutines since ingestion is mostly I/O bound, making it easier to
// perform the computational work concurrently.
for range ingestConcurrency {
group.GoCtx(func(ctx context.Context) error {
err := ib.ingestIndexEntries(ctx, indexEntriesCh, progCh)
if err != nil {
return errors.Wrap(err, "failed to ingest index entries during backfill")
}
return nil
})
}

if err := group.Wait(); err != nil {
return err
Expand Down
110 changes: 0 additions & 110 deletions pkg/sql/schema_changer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1600,116 +1600,6 @@ CREATE TABLE t.test (k INT PRIMARY KEY, v INT);
}
}

// Test schema change purge failure doesn't leave DB in a bad state.
func TestSchemaChangePurgeFailure(t *testing.T) {
defer leaktest.AfterTest(t)()
defer log.Scope(t).Close(t)

for _, schemaChangerSetup := range []string{
"SET use_declarative_schema_changer='off'",
"SET use_declarative_schema_changer='on'",
} {
params, _ := createTestServerParams()
const chunkSize = 200

var getKeyCount func() (int, error)
countBeforeRollback := 0
params.Knobs = base.TestingKnobs{
SQLSchemaChanger: &sql.SchemaChangerTestingKnobs{
RunBeforeOnFailOrCancel: func(jobID jobspb.JobID) error {
cnt, err := getKeyCount()
if err != nil {
return err
}
countBeforeRollback = cnt
return nil
},
},
DistSQL: &execinfra.TestingKnobs{
BulkAdderFlushesEveryBatch: true,
},
SQLDeclarativeSchemaChanger: &scexec.TestingKnobs{
RunBeforeMakingPostCommitPlan: func(inRollback bool) error {
if inRollback {
cnt, err := getKeyCount()
if err != nil {
return err
}
countBeforeRollback = cnt
}
return nil
},
},
}
server, sqlDB, kvDB := serverutils.StartServer(t, params)
defer server.Stopper().Stop(context.Background())
codec := server.ApplicationLayer().Codec()

_, err := sqlDB.Exec(fmt.Sprintf("SET CLUSTER SETTING bulkio.index_backfill.batch_size = %d", chunkSize))
require.NoError(t, err)

getKeyCount = func() (int, error) {
return sqltestutils.GetTableKeyCount(ctx, kvDB, codec)
}
// Disable strict GC TTL enforcement because we're going to shove a zero-value
// TTL into the system with AddImmediateGCZoneConfig.
defer sqltestutils.DisableGCTTLStrictEnforcement(t, sqlDB)()

_, err = sqlDB.Exec(schemaChangerSetup)
require.NoError(t, err)

if _, err := sqlDB.Exec(`
CREATE DATABASE t;
CREATE TABLE t.test (k INT PRIMARY KEY, v INT);
`); err != nil {
t.Fatal(err)
}

// Bulk insert.
const maxValue = chunkSize + 1
if err := sqltestutils.BulkInsertIntoTable(sqlDB, maxValue); err != nil {
t.Fatal(err)
}

// Add a row with a duplicate value=0 which is the same
// value as for the key maxValue.
if _, err := sqlDB.Exec(
`INSERT INTO t.test VALUES ($1, $2)`, maxValue+1, 0,
); err != nil {
t.Fatal(err)
}

// A schema change that violates integrity constraints.
if _, err := sqlDB.Exec(
"CREATE UNIQUE INDEX foo ON t.test (v)",
); !testutils.IsError(err, `violates unique constraint "foo"`) {
t.Fatal(err)
}

// The index doesn't exist
if _, err := sqlDB.Query(
`SELECT v from t.test@foo`,
); !testutils.IsError(err, "index .* not found") {
t.Fatal(err)
}

// countBeforeRollback is assigned in the rollback testing knob which is
// called before rollback starts so that the first chunk (200 keys) written
// is still visible. The first chunk is visible because there is no
// duplicate keys within it. The duplicate keys only exist in the second
// chunk. Also note that we wrote maxValue+1 rows, and there is 1 extra key
// from kv.
require.Equal(t, countBeforeRollback, maxValue+2+chunkSize)

// No garbage left behind after rollback. This check should succeed pretty
// fast since we use `DelRange` in GC and `CheckTableKeyCount` cannot see
// tombstones.
testutils.SucceedsSoon(t, func() error {
return sqltestutils.CheckTableKeyCount(ctx, kvDB, codec, 1, maxValue+1)
})
}
}

// Test schema change failure after a backfill checkpoint has been written
// doesn't leave the DB in a bad state.
func TestSchemaChangeFailureAfterCheckpointing(t *testing.T) {
Expand Down