release-24.2: colexec: fix type schema corruption in an edge case

pkg/sql/colexec/colbuilder/execplan.go

@@ -720,6 +720,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);