Merge #126280

126280: crosscluster/logical: use fresh internal session data r=yuzefovich a=yuzefovich

This commit switches usage of the current session's data in favor of a fresh internal session data when instantiating the executor and the txn used by the LWW row processor. I didn't quite tracked down why, but it appears that current session's data (which comes from the JobExecCtx) is inconsistent (in some cases I observed it to use Go defaults, perhaps because it was being mutated (?)), so using the fresh internal session data would alleviate that. (This is also exactly what would happen if we didn't use the WithSessionData option, but this way avoid an allocation on each executor usage). Microbenchmark show no difference (which is a good thing).

Epic: None

Release note: None

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

Author: craig[bot]

pkg/ccl/crosscluster/logical/BUILD.bazel

@@ -99,7 +99,6 @@ go_test(
         "//pkg/sql/randgen",
         "//pkg/sql/rowenc",
         "//pkg/sql/sem/tree",
-        "//pkg/sql/sessiondata",
         "//pkg/testutils",
         "//pkg/testutils/jobutils",
         "//pkg/testutils/serverutils",

pkg/ccl/crosscluster/logical/logical_replication_writer_processor.go

@@ -120,9 +120,9 @@ func newLogicalReplicationWriterProcessor(
 	for i := range bhPool {
 		rp, err := makeSQLLastWriteWinsHandler(
 			ctx, flowCtx.Cfg.Settings, spec.TableDescriptors,
-			// Initialize the executor with the copy of the current session's
-			// variables in order to avoid creating a fresh copy on each usage.
-			flowCtx.Cfg.DB.Executor(isql.WithSessionData(flowCtx.EvalCtx.SessionData().Clone())),
+			// Initialize the executor with a fresh session data - this will
+			// avoid creating a new copy on each executor usage.
+			flowCtx.Cfg.DB.Executor(isql.WithSessionData(sql.NewInternalSessionData(ctx, flowCtx.Cfg.Settings, "" /* opName */))),
 		)
 		if err != nil {
 			return nil, err
@@ -131,7 +131,7 @@ func newLogicalReplicationWriterProcessor(
 			db:       flowCtx.Cfg.DB,
 			rp:       rp,
 			settings: flowCtx.Cfg.Settings,
-			sd:       flowCtx.EvalCtx.SessionData().Clone(),
+			sd:       sql.NewInternalSessionData(ctx, flowCtx.Cfg.Settings, "" /* opName */),
 		}
 	}
 

pkg/ccl/crosscluster/logical/lww_row_processor_test.go

@@ -16,10 +16,10 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/ccl/crosscluster/replicationtestutils"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
-	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
@@ -126,14 +126,8 @@ func BenchmarkLWWInsertBatch(b *testing.B) {
 	runner.Exec(b, "ALTER TABLE tab ADD COLUMN crdb_internal_origin_timestamp DECIMAL NOT VISIBLE DEFAULT NULL ON UPDATE NULL")
 
 	desc := desctestutils.TestingGetPublicTableDescriptor(kvDB, s.Codec(), "defaultdb", tableName)
-	// Simulate the setup of the main code path where we have access to the
-	// session data defaults.
-	var sd *sessiondata.SessionData
-	err := s.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
-		sd = txn.SessionData().Clone()
-		return nil
-	})
-	require.NoError(b, err)
+	// Simulate how we set up the row processor on the main code path.
+	sd := sql.NewInternalSessionData(ctx, s.ClusterSettings(), "" /* opName */)
 	rp, err := makeSQLLastWriteWinsHandler(ctx, s.ClusterSettings(), map[int32]descpb.TableDescriptor{
 		int32(desc.GetID()): *desc.TableDesc(),
 	}, s.InternalDB().(isql.DB).Executor(isql.WithSessionData(sd)))