Merge #158339

158339: sql/bulkingest: add IngestFileProcessor for SST ingestion r=spilchen a=spilchen

This introduces the IngestFileProcessor, a new DistSQL porcessor that ingests SST files directly into the KV layer. This is part of the new distributed merge pipeline. The SSTs will be provided as input from an earlier stage in the pipeline (merge processor).

The processor is registered but not yet used by any SQL operations. Future work will integrate it into the distribute merge pipeline.

Epic: CRDB-48845
Fixes: #156659
Release note: none

Co-authored by: `@jeffswenson` 

Co-authored-by: Matt Spilchen <[email protected]>

Author: craig[bot]

pkg/cloud/nodelocal/nodelocal_storage.go

@@ -110,9 +110,6 @@ func makeLocalFileStorage(
 		return nil, errors.New("nodelocal storage is not available")
 	}
 	cfg := dest.LocalFileConfig
-	if cfg.Path == "" {
-		return nil, errors.Errorf("local storage requested but path not provided")
-	}
 	client, err := args.BlobClientFactory(ctx, cfg.NodeID)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create blob client")

pkg/sql/bulkingest/BUILD.bazel

@@ -3,25 +3,47 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "bulkingest",
     srcs = [
+        "ingest.go",
+        "ingest_file_processor.go",
+        "plan_ingest.go",
         "split.go",
         "split_picker.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/bulkingest",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/cloud",
+        "//pkg/clusterversion",
         "//pkg/keys",
         "//pkg/kv",
         "//pkg/roachpb",
+        "//pkg/settings",
+        "//pkg/sql",
+        "//pkg/sql/bulkutil",
+        "//pkg/sql/execinfra",
         "//pkg/sql/execinfrapb",
+        "//pkg/sql/physicalplan",
+        "//pkg/sql/rowexec",
+        "//pkg/sql/sem/tree",
+        "//pkg/sql/types",
+        "//pkg/util/ctxgroup",
+        "//pkg/util/errorutil/unimplemented",
+        "//pkg/util/hlc",
+        "//pkg/util/ioctx",
         "//pkg/util/log",
+        "//pkg/util/mon",
+        "//pkg/util/tracing",
         "@com_github_cockroachdb_errors//:errors",
     ],
 )
 
 go_test(
     name = "bulkingest_test",
     srcs = [
+        "ingest_file_processor_test.go",
+        "ingest_test.go",
         "main_test.go",
+        "plan_ingest_test.go",
         "split_picker_test.go",
         "split_test.go",
     ],
@@ -32,14 +54,26 @@ go_test(
     }),
     deps = [
         "//pkg/base",
+        "//pkg/cloud/externalconn/providers",
         "//pkg/keys",
         "//pkg/kv/kvclient/kvtenant",
+        "//pkg/kv/kvpb",
+        "//pkg/kv/kvserver",
         "//pkg/roachpb",
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",
+        "//pkg/security/username",
         "//pkg/server",
+        "//pkg/settings/cluster",
+        "//pkg/sql",
+        "//pkg/sql/bulksst",
+        "//pkg/sql/catalog",
         "//pkg/sql/catalog/descpb",
+        "//pkg/sql/catalog/desctestutils",
         "//pkg/sql/execinfrapb",
+        "//pkg/sql/rowenc",
+        "//pkg/storage",
+        "//pkg/testutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",

pkg/sql/bulkingest/ingest.go

@@ -0,0 +1,61 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package bulkingest
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+)
+
+// IngestFiles runs the distributed ingest phase for a set of pre-built SSTs.
+// It:
+//   - picks range splits that align with SST boundaries,
+//   - pre-splits and scatters those ranges, and
+//   - executes an ingest DistSQL flow that AddSSTables each file.
+func IngestFiles(
+	ctx context.Context,
+	execCtx sql.JobExecContext,
+	spans []roachpb.Span,
+	ssts []execinfrapb.BulkMergeSpec_SST,
+) error {
+	splits, err := pickSplits(spans, ssts)
+	if err != nil {
+		return err
+	}
+
+	db := execCtx.ExecCfg().InternalDB.KV()
+	if err := splitAndScatterSpans(ctx, db, splits); err != nil {
+		return err
+	}
+
+	plan, planCtx, err := planIngest(ctx, execCtx, ssts)
+	if err != nil {
+		return err
+	}
+
+	res := sql.NewMetadataOnlyMetadataCallbackWriter(func(context.Context, *execinfrapb.ProducerMetadata) error { return nil })
+	recv := sql.MakeDistSQLReceiver(
+		ctx,
+		res,
+		tree.Ack,
+		execCtx.ExecCfg().RangeDescriptorCache, /* rangeCache */
+		nil,                                    /* txn - the flow does not read or write the database */
+		nil,                                    /* clockUpdater */
+		execCtx.ExtendedEvalContext().Tracing,
+	)
+	defer recv.Release()
+
+	evalCtxCopy := execCtx.ExtendedEvalContext().Context.Copy()
+	execCtx.DistSQLPlanner().Run(
+		ctx, planCtx, nil, plan, recv, evalCtxCopy, nil, /* finishedSetupFn */
+	)
+
+	return res.Err()
+}