Merge cockroachdb#133624

133624: colexec: fix type schema corruption in an edge case r=yuzefovich a=yuzefovich

This commit fixes type schema corruption in the vectorized engine in an edge case. In particular, consider the following circumstances:
- during the physical planning, when creating a new stage of processors, we often reuse the same type slice (stored in
`InputSyncSpec.ColumnTypes`) that we get from the previous stage. In other words, we might have memory aliasing, but only on the gateway node because the remote nodes get their specs deserialized and each has its own memory allocation.
- throughout the vectorized operator planning, as of 85fd4fb, for each newly projected vector we append the corresponding type to the type slice we have in scope. We also capture intermediate state of the type slice by some operators (e.g. `BatchSchemaSubsetEnforcer`).
- as expected, when appending a type to the slice, if there is enough capacity, we reuse it, meaning that we often append to the slice that came to us via `InputSyncSpec.ColumnTypes`.
- now, if we have two stages of processors that happened to share the same underlying type slice with some free capacity AND we needed to append vectors for each stage, then we might corrupt the type schema captured by an operator for the earlier stage when performing vectorized planning for the later stage.

The bug is effectively the same as the comment deleted by 85fd4fb outlined:
```
// As an example, consider the following scenario in the context of
// planFilterExpr method:
// 1. r.ColumnTypes={types.Bool} with len=1 and cap=4
// 2. planSelectionOperators adds another types.Int column, so
//    filterColumnTypes={types.Bool, types.Int} with len=2 and cap=4
//    Crucially, it uses exact same underlying array as r.ColumnTypes
//    uses.
// 3. we project out second column, so r.ColumnTypes={types.Bool}
// 4. later, we add another types.Float column, so
//    r.ColumnTypes={types.Bool, types.Float}, but there is enough
//    capacity in the array, so we simply overwrite the second slot
//    with the new type which corrupts filterColumnTypes to become
//    {types.Bool, types.Float}, and we can get into a runtime type
//    mismatch situation.
```
The only differences are:
- aliasing of the type slice occurs via the `InputSyncSpec.ColumnTypes` that is often used as the starting points for populating `NewColOperatorResult.ColumnTypes` which is used throughout the vectorized operator planning
- columns are "projected out" by sharing the type schema between two stages of DistSQL processors.

This commit addresses this issue by capping the slice to its length right before we get into the vectorized planning. This will make it so that if we need to append a type, then we'll make a fresh allocation, and any possible memory aliasing with a different stage of processors will be gone.

I haven't quite figured out the exact conditions that are needed for this bug to occur, but my intuition says that it should be quite rare in practice (otherwise we'd have seen this much sooner given that the offending commit was merged more than a year ago and was backported to older branches).

Fixes: cockroachdb#130402.

Release note (bug fix): Previously, CockroachDB could encounter an internal error of the form `interface conversion: coldata.Column is` in an edge case and this is now fixed. The bug is present in versions 22.2.13+, 23.1.9+, 23.2+.

Co-authored-by: Yahor Yuzefovich <[email protected]>

pkg/sql/colexec/colbuilder/execplan.go

@@ -727,6 +727,25 @@ func NewColOperator(
 	result := opResult{NewColOperatorResult: colexecargs.GetNewColOperatorResult()}
 	r := result.NewColOperatorResult
 	spec := args.Spec
+	// Throughout this method we often use the type slice from the input spec to
+	// create the type schema of an operator. However, it is possible that the
+	// same type slice is shared by multiple stages of processors. If it just so
+	// happens that there is free capacity in the slice, and we append to it
+	// when planning operators for both stages, we might corrupt the type schema
+	// captured by the operators for the earlier stage. In order to prevent such
+	// type schema corruption we cap the slice to force creation of a fresh copy
+	// on the first append.
+	if flowCtx.Gateway {
+		// Sharing of the same type slice is only possible on the gateway node
+		// because we don't serialize the specs created during the physical
+		// planning. On the remote nodes each stage of processors gets their own
+		// allocation, so there is no aliasing that can lead to the type schema
+		// corruption.
+		for i := range spec.Input {
+			inputSpec := &spec.Input[i]
+			inputSpec.ColumnTypes = inputSpec.ColumnTypes[:len(inputSpec.ColumnTypes):len(inputSpec.ColumnTypes)]
+		}
+	}
 	inputs := args.Inputs
 	if args.Factory == nil {
 		// This code path is only used in tests.

pkg/sql/logictest/testdata/logic_test/vectorize_types

@@ -182,3 +182,39 @@ FROM
 	t107615
 WHERE
 	(_bool OR (NOT _bool));
+
+# Regression test for corrupting the column type schema captured by a vectorized
+# operator due to sharing the same type slice by multiple stages of processors
+# during physical planning (#130402).
+statement ok
+CREATE TABLE abcdef (
+  a STRING,
+  b INT4,
+  c INT4,
+  d INT4,
+  e STRING,
+  f STRING
+);
+INSERT INTO abcdef (a, b, c, d, e, f) VALUES ('a', 0, 0, 0, 'e', 'f');
+CREATE TABLE ghijkl (
+  g INT4,
+  h INT4,
+  i BOOL,
+  j INT4,
+  k STRING,
+  l STRING
+);
+INSERT INTO ghijkl (g, h, i, j, k, l) VALUES (0, 0, true, -1, 'k', 'l');
+SELECT c7, c1 >= (SELECT c FROM abcdef WHERE e != c3), c5
+FROM (
+  SELECT
+    generate_series(g, d) AS c0,
+    h AS c1,
+    l AS c3,
+    k AS c5,
+    c AS c7
+  FROM ghijkl
+  LEFT JOIN abcdef ON b > j OR i NOT IN (e NOT IN (f, a),)
+  LIMIT 2
+)
+WHERE c0 NOT IN (SELECT NULL FROM ghijkl);