Merge #60550 #60748

60550: sql: add nodes for each EXPLAIN ANALYZE operator r=RaduBerinde a=RaduBerinde

Show the cluster nodes involved in the execution of each operator.

Note that this info does not show up in the non-analyze EXPLAIN. It is
technically much more challenging to do that because of the indirect
way we do distsql planning. Once we have the new DistSQL exec factory,
we will be able to add it.

Note on the implementation: I started by trying to make execution set
`ComponentID.NodeID` in all cases, but I got stuck in `ProcessorBase`
where we only have a `SQLIDContainer` available. I don't fully
understand the new abstraction and whether the distsql components and
flows should really use SQLIDs instead of NodeIDs.

Unfortunately, there is not much we can do to test this currently
(other than manual testing). I will investigate making the
"deterministic" flag more fine-grained, so that we can hide truly
non-deterministic values (like timings) separately from those that
just vary with the configuration (query distribution).

Example:
```
movr> EXPLAIN ANALYZE SELECT * FROM rides JOIN vehicles ON rides.vehicle_id = vehicles.id;
                    info
--------------------------------------------
  planning time: 158µs
  execution time: 7ms
  distribution: full
  vectorized: true

   hash join
  │ cluster nodes: n1, n2, n3
  │ actual row count: 500
  │ equality: (vehicle_id) = (id)
  │
  ├──  scan
  │     cluster nodes: n1, n2, n3
  │     actual row count: 500
  │     KV rows read: 500
  │     KV bytes read: 86 KiB
  │     missing stats
  │     table: rides@primary
  │     spans: FULL SCAN
  │
  └──  scan
        cluster nodes: n1, n2, n3
        actual row count: 15
        KV rows read: 15
        KV bytes read: 2.3 KiB
        missing stats
        table: vehicles@primary
        spans: FULL SCAN
```

Release note (sql change): EXPLAIN ANALYZE now includes the nodes
which were involved in the execution of each operator in the tree.

Informs #59860.

60748: sql: remove experimental setting for virtual columns r=RaduBerinde a=RaduBerinde

Informs #57608.

Release note (sql change): CockroachDB now supports VIRTUAL computed
columns (as opposed to STORED). These are computed columns that are
not stored in the primary index and are recomputed as necessary.

Co-authored-by: Radu Berinde <[email protected]>

pkg/sql/add_column.go

@@ -20,7 +20,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlerrors"
-	"github.com/cockroachdb/cockroach/pkg/util/errorutil/unimplemented"
 	"github.com/cockroachdb/errors"
 )
 
@@ -128,9 +127,6 @@ func (p *planner) addColumnImpl(
 	}
 
 	if d.IsComputed() {
-		if d.IsVirtual() && !sessionData.VirtualColumnsEnabled {
-			return unimplemented.NewWithIssue(57608, "virtual computed columns")
-		}
 		computedColValidator := schemaexpr.MakeComputedColumnValidator(
 			params.ctx,
 			n.tableDesc,

pkg/sql/colflow/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/colflow",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/base",
         "//pkg/col/coldata",
         "//pkg/col/coldataext",
         "//pkg/rpc/nodedialer",

pkg/sql/colflow/vectorized_flow.go

@@ -20,6 +20,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/col/coldata"
 	"github.com/cockroachdb/cockroach/pkg/col/coldataext"
 	"github.com/cockroachdb/cockroach/pkg/rpc/nodedialer"
@@ -831,9 +832,12 @@ func (s *vectorizedFlowCreator) setupInput(
 			s.addStreamEndpoint(inputStream.StreamID, inbox, s.waitGroup)
 			op := colexecbase.Operator(inbox)
 			if s.recordingStats {
-				op, err = s.wrapWithNetworkVectorizedStatsCollector(
-					inbox, flowCtx.StreamComponentID(inputStream.StreamID), latency,
+				// Note: we can't use flowCtx.StreamComponentID because the stream does
+				// not originate from this node (we are the target node).
+				compID := execinfrapb.StreamComponentID(
+					base.SQLInstanceID(inputStream.OriginNodeID), flowCtx.ID, inputStream.StreamID,
 				)
+				op, err = s.wrapWithNetworkVectorizedStatsCollector(inbox, compID, latency)
 				if err != nil {
 					return nil, nil, nil, err
 				}
@@ -958,7 +962,7 @@ func (s *vectorizedFlowCreator) setupOutput(
 							// flow-level span.
 							span.SetTag(execinfrapb.FlowIDTagKey, flowCtx.ID)
 							span.SetSpanStats(&execinfrapb.ComponentStats{
-								Component: execinfrapb.FlowComponentID(outputStream.OriginNodeID, flowCtx.ID),
+								Component: execinfrapb.FlowComponentID(base.SQLInstanceID(outputStream.OriginNodeID), flowCtx.ID),
 								FlowStats: execinfrapb.FlowStats{
 									MaxMemUsage: optional.MakeUint(uint64(flowCtx.EvalCtx.Mon.MaximumBytes())),
 								},

pkg/sql/create_table.go

@@ -1704,9 +1704,6 @@ func NewTableDesc(
 				columnDefaultExprs = append(columnDefaultExprs, nil)
 			}
 			if d.IsVirtual() {
-				if !sessionData.VirtualColumnsEnabled {
-					return nil, unimplemented.NewWithIssue(57608, "virtual computed columns")
-				}
 				if !evalCtx.Settings.Version.IsActive(ctx, clusterversion.VirtualComputedColumns) {
 					return nil, pgerror.Newf(pgcode.FeatureNotSupported,
 						"version %v must be finalized to use virtual columns",

pkg/sql/distsql/inbound_test.go

@@ -92,6 +92,7 @@ func TestOutboxInboundStreamIntegration(t *testing.T) {
 			NodeDialer: nodedialer.New(rpcContext, staticAddressResolver(ln.Addr())),
 			Stopper:    outboxStopper,
 		},
+		NodeID: base.TestingIDContainer,
 	}
 
 	streamID := execinfrapb.StreamID(1)

pkg/sql/distsql_physical_planner.go

@@ -16,6 +16,7 @@ import (
 	"reflect"
 	"sort"
 
+	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/gossip"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/keys"
@@ -2742,10 +2743,11 @@ func (dsp *DistSQLPlanner) createPhysPlanForPlanNode(
 
 	if planCtx.traceMetadata != nil {
 		processors := make(execComponents, len(plan.ResultRouters))
-		for i := range plan.ResultRouters {
+		for i, resultProcIdx := range plan.ResultRouters {
 			processors[i] = execinfrapb.ProcessorComponentID(
+				base.SQLInstanceID(plan.Processors[resultProcIdx].Node),
 				execinfrapb.FlowID{UUID: planCtx.infra.FlowID},
-				int32(plan.ResultRouters[i]),
+				int32(resultProcIdx),
 			)
 		}
 		planCtx.traceMetadata.associateNodeWithComponents(node, processors)

pkg/sql/exec_util.go

@@ -2276,11 +2276,6 @@ func (m *sessionDataMutator) SetAlterColumnTypeGeneral(val bool) {
 	m.data.AlterColumnTypeGeneralEnabled = val
 }
 
-// TODO(radu): remove this once the feature is stable.
-func (m *sessionDataMutator) SetVirtualColumnsEnabled(val bool) {
-	m.data.VirtualColumnsEnabled = val
-}
-
 // TODO(rytaft): remove this once unique without index constraints are fully
 // supported.
 func (m *sessionDataMutator) SetUniqueWithoutIndexConstraints(val bool) {

pkg/sql/execinfra/flow_context.go

@@ -108,10 +108,11 @@ func (ctx *FlowCtx) Codec() keys.SQLCodec {
 // ProcessorComponentID returns a ComponentID for the given processor in this
 // flow.
 func (ctx *FlowCtx) ProcessorComponentID(procID int32) execinfrapb.ComponentID {
-	return execinfrapb.ProcessorComponentID(ctx.ID, procID)
+	return execinfrapb.ProcessorComponentID(ctx.NodeID.SQLInstanceID(), ctx.ID, procID)
 }
 
 // StreamComponentID returns a ComponentID for the given stream in this flow.
+// The stream must originate from the node associated with this FlowCtx.
 func (ctx *FlowCtx) StreamComponentID(streamID execinfrapb.StreamID) execinfrapb.ComponentID {
-	return execinfrapb.StreamComponentID(ctx.ID, streamID)
+	return execinfrapb.StreamComponentID(ctx.NodeID.SQLInstanceID(), ctx.ID, streamID)
 }

pkg/sql/execinfrapb/component_stats.go

@@ -15,7 +15,7 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/util/humanizeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
@@ -26,29 +26,35 @@ import (
 )
 
 // ProcessorComponentID returns a ComponentID for the given processor in a flow.
-func ProcessorComponentID(flowID FlowID, processorID int32) ComponentID {
+func ProcessorComponentID(
+	instanceID base.SQLInstanceID, flowID FlowID, processorID int32,
+) ComponentID {
 	return ComponentID{
-		FlowID: flowID,
-		Type:   ComponentID_PROCESSOR,
-		ID:     processorID,
+		FlowID:        flowID,
+		Type:          ComponentID_PROCESSOR,
+		ID:            processorID,
+		SQLInstanceID: instanceID,
 	}
 }
 
 // StreamComponentID returns a ComponentID for the given stream in a flow.
-func StreamComponentID(flowID FlowID, streamID StreamID) ComponentID {
+func StreamComponentID(
+	originInstanceID base.SQLInstanceID, flowID FlowID, streamID StreamID,
+) ComponentID {
 	return ComponentID{
-		FlowID: flowID,
-		Type:   ComponentID_STREAM,
-		ID:     int32(streamID),
+		FlowID:        flowID,
+		Type:          ComponentID_STREAM,
+		ID:            int32(streamID),
+		SQLInstanceID: originInstanceID,
 	}
 }
 
 // FlowComponentID returns a ComponentID for the given flow.
-func FlowComponentID(nodeID roachpb.NodeID, flowID FlowID) ComponentID {
+func FlowComponentID(instanceID base.SQLInstanceID, flowID FlowID) ComponentID {
 	return ComponentID{
-		FlowID: flowID,
-		Type:   ComponentID_FLOW,
-		NodeID: nodeID,
+		FlowID:        flowID,
+		Type:          ComponentID_FLOW,
+		SQLInstanceID: instanceID,
 	}
 }