Merge #143668 #143776

143668: jsonpath: add support for `() is unknown` r=normanchenn a=normanchenn

#### jsonpath: extract boolean operations into `evalBoolean`

This commit moves the switch statement evaluation of boolean returning
operators into a separate `evalBoolean`. This allows for other operators
that expect to return a boolean (logical AND/OR/NOT, filters) to do so
without needing to do `isBool` checks after an `eval` call.

Epic: None
Release note: None

#### jsonpath: add support for `() is unknown`

This commit adds support for using `() is unknown` within jsonpath
queries. This returns a boolean - whether or not the enclosed predicate
returned unknown.

Epic: None
Release note (sql change): Add support for `() is unknown` in JSONPath
queries. For example, `SELECT jsonb_path_query('{}', '($ < 1) is unknown');`.


143776: logical: use sql writer for immediate mode r=jeffswenson a=jeffswenson

This changes the default immediate mode writer to the sql writer. The logical_replication.consumer.immediate_mode_writer setting may be used to switch back to the 'legacy-kv' writer.

The SQL writer will now always use batching. Previously, batching was
disabled if the table did not include any indexes, but that seems to be
a hold over from when the kv writer used batching.

Release note: none
Part of: #143754

Co-authored-by: Norman Chen <[email protected]>
Co-authored-by: Jeff Swenson <[email protected]>

Author: 3 people

pkg/crosscluster/logical/BUILD.bazel

@@ -135,6 +135,7 @@ go_test(
         "//pkg/ccl/changefeedccl/cdctest",
         "//pkg/ccl/changefeedccl/changefeedbase",
         "//pkg/ccl/storageccl",
+        "//pkg/clusterversion",
         "//pkg/crosscluster/replicationtestutils",
         "//pkg/crosscluster/streamclient",
         "//pkg/crosscluster/streamclient/randclient",
@@ -149,6 +150,7 @@ go_test(
         "//pkg/security/securitytest",
         "//pkg/security/username",
         "//pkg/server",
+        "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/catpb",

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach-go/v2/crdb"
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcevent"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/replicationtestutils"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/streamclient"
 	_ "github.com/cockroachdb/cockroach/pkg/crosscluster/streamclient/randclient"
@@ -31,6 +32,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
@@ -1689,7 +1691,8 @@ func GetReverseJobID(
 }
 
 type mockBatchHandler struct {
-	err error
+	err       error
+	batchSize int
 }
 
 var _ BatchHandler = &mockBatchHandler{}
@@ -1707,6 +1710,7 @@ func (m mockBatchHandler) SetSyntheticFailurePercent(_ uint32) {}
 func (m mockBatchHandler) Close(context.Context)               {}
 func (m mockBatchHandler) ReportMutations(_ *stats.Refresher)  {}
 func (m mockBatchHandler) ReleaseLeases(_ context.Context)     {}
+func (m mockBatchHandler) BatchSize() int                      { return m.batchSize }
 
 type mockDLQ int
 
@@ -1734,16 +1738,19 @@ func TestFlushErrorHandling(t *testing.T) {
 	ctx := context.Background()
 	dlq := mockDLQ(0)
 	lrw := &logicalReplicationWriterProcessor{
-		metrics:      MakeMetrics(0).(*Metrics),
-		getBatchSize: func() int { return 1 },
-		dlqClient:    &dlq,
+		metrics:   MakeMetrics(0).(*Metrics),
+		dlqClient: &dlq,
 	}
 	lrw.purgatory.flush = lrw.flushBuffer
 	lrw.purgatory.bytesGauge = lrw.metrics.RetryQueueBytes
 	lrw.purgatory.eventsGauge = lrw.metrics.RetryQueueEvents
 	lrw.purgatory.debug = &streampb.DebugLogicalConsumerStatus{}
 
-	lrw.bh = []BatchHandler{(mockBatchHandler{pgerror.New(pgcode.UniqueViolation, "index write conflict")})}
+	lrw.bh = []BatchHandler{(mockBatchHandler{
+		err:       pgerror.New(pgcode.UniqueViolation, "index write conflict"),
+		batchSize: 1,
+	})}
+
 	lrw.bhStats = make([]flushStats, 1)
 
 	lrw.purgatory.byteLimit = func() int64 { return 1 }
@@ -2693,3 +2700,43 @@ func TestFailDestAfterSourceFailure(t *testing.T) {
 	dbB.Exec(t, "RESUME JOB $1", jobBID)
 	jobutils.WaitForJobToFail(t, dbB, jobBID)
 }
+
+func TestGetWriterType(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	ctx := context.Background()
+
+	t.Run("validated-mode", func(t *testing.T) {
+		st := cluster.MakeTestingClusterSettings()
+		wt, err := getWriterType(ctx, jobspb.LogicalReplicationDetails_Validated, st)
+		require.NoError(t, err)
+		require.Equal(t, writerTypeSQL, wt)
+	})
+
+	t.Run("immediate-mode-pre-25.2", func(t *testing.T) {
+		st := cluster.MakeTestingClusterSettingsWithVersions(
+			clusterversion.V25_1.Version(),
+			clusterversion.V25_1.Version(),
+			true, /* initializeVersion */
+		)
+		wt, err := getWriterType(ctx, jobspb.LogicalReplicationDetails_Immediate, st)
+		require.NoError(t, err)
+		require.Equal(t, writerTypeSQL, wt)
+	})
+
+	t.Run("immediate-mode-post-25.2", func(t *testing.T) {
+		st := cluster.MakeTestingClusterSettingsWithVersions(
+			clusterversion.V25_2.Version(),
+			clusterversion.PreviousRelease.Version(),
+			true, /* initializeVersion */
+		)
+		immediateModeWriter.Override(ctx, &st.SV, string(writerTypeSQL))
+		wt, err := getWriterType(ctx, jobspb.LogicalReplicationDetails_Immediate, st)
+		require.NoError(t, err)
+		require.Equal(t, writerTypeSQL, wt)
+
+		immediateModeWriter.Override(ctx, &st.SV, string(writerTypeLegacyKV))
+		wt, err = getWriterType(ctx, jobspb.LogicalReplicationDetails_Immediate, st)
+		require.NoError(t, err)
+		require.Equal(t, writerTypeLegacyKV, wt)
+	})
+}

pkg/crosscluster/logical/logical_replication_writer_processor.go

@@ -16,13 +16,15 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcevent"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/streamclient"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
@@ -93,6 +95,30 @@ var maxChunkSize = settings.RegisterIntSetting(
 	settings.NonNegativeInt,
 )
 
+type writerType string
+
+const (
+	// writerTypeSQL uses the SQL layer to write replicated rows.
+	writerTypeSQL writerType = "sql"
+	// writerTypeLegacyKV uses the legacy KV layer to write rows. The KV writer
+	// is deprecated because it does not support the full set of features of the
+	// SQL writer.
+	writerTypeLegacyKV writerType = "legacy-kv"
+)
+
+var immediateModeWriter = settings.RegisterStringSetting(
+	settings.ApplicationLevel,
+	"logical_replication.consumer.immediate_mode_writer",
+	"the writer to use when in immediate mode",
+	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(writerTypeSQL), string(writerTypeLegacyKV)),
+	settings.WithValidateString(func(sv *settings.Values, val string) error {
+		if val != string(writerTypeSQL) && val != string(writerTypeLegacyKV) {
+			return errors.Newf("immediate mode writer must be either 'sql' or 'legacy-kv', got '%s'", val)
+		}
+		return nil
+	}),
+)
+
 // logicalReplicationWriterProcessor consumes a cross-cluster replication stream
 // by decoding kvs in it to logical changes and applying them by executing DMLs.
 type logicalReplicationWriterProcessor struct {
@@ -106,8 +132,6 @@ type logicalReplicationWriterProcessor struct {
 
 	configByTable map[descpb.ID]sqlProcessorTableConfig
 
-	getBatchSize func() int
-
 	streamPartitionClient streamclient.Client
 
 	// frontier keeps track of the progress for the spans tracked by this processor
@@ -211,40 +235,9 @@ func newLogicalReplicationWriterProcessor(
 	}
 
 	lrw := &logicalReplicationWriterProcessor{
-		configByTable: procConfigByDestTableID,
-		spec:          spec,
-		processorID:   processorID,
-		getBatchSize: func() int {
-			// TODO(ssd): We set this to 1 since putting more than 1
-			// row in a KV batch using the new ConditionalPut-based
-			// conflict resolution would require more complex error
-			// handling and tracking that we haven't implemented
-			// yet.
-			if spec.Mode == jobspb.LogicalReplicationDetails_Immediate {
-				return 1
-			}
-			// We want to decide whether to use implicit txns or not based on
-			// the schema of the dest table. Benchmarking has shown that
-			// implicit txns are beneficial on tables with no secondary indexes
-			// whereas explicit txns are beneficial when at least one secondary
-			// index is present.
-			//
-			// Unfortunately, if we have multiple replication pairs, we don't
-			// know which tables will be affected by this batch before deciding
-			// on the batch size, so we'll use a heuristic such that we'll use
-			// the implicit txns if at least half of the dest tables are
-			// without the secondary indexes. If we only have a single
-			// replication pair, then this heuristic gives us the precise
-			// recommendation.
-			//
-			// (Here we have access to the descriptor of the source table, but
-			// for now we assume that the source and the dest descriptors are
-			// similar.)
-			if 2*numTablesWithSecondaryIndexes < len(procConfigByDestTableID) && useImplicitTxns.Get(&flowCtx.Cfg.Settings.SV) {
-				return 1
-			}
-			return int(flushBatchSize.Get(&flowCtx.Cfg.Settings.SV))
-		},
+		configByTable:  procConfigByDestTableID,
+		spec:           spec,
+		processorID:    processorID,
 		frontier:       frontier,
 		stopCh:         make(chan struct{}),
 		checkpointCh:   make(chan []jobspb.ResolvedSpan),
@@ -725,14 +718,12 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 		var err error
 		sd := sql.NewInternalSessionData(ctx, flowCtx.Cfg.Settings, "" /* opName */)
 
-		if lrw.spec.Mode == jobspb.LogicalReplicationDetails_Immediate {
-			evalCtx := flowCtx.NewEvalCtx()
-			evalCtx.SessionDataStack.Push(sd)
-			rp, err = newKVRowProcessor(ctx, flowCtx.Cfg, evalCtx, lrw.spec, lrw.configByTable)
-			if err != nil {
-				return err
-			}
-		} else {
+		writer, err := getWriterType(ctx, lrw.spec.Mode, flowCtx.Cfg.Settings)
+		if err != nil {
+			return err
+		}
+		switch writer {
+		case writerTypeSQL:
 			rp, err = makeSQLProcessor(
 				ctx, flowCtx.Cfg.Settings, lrw.configByTable,
 				jobspb.JobID(lrw.spec.JobID),
@@ -747,6 +738,13 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 			if err != nil {
 				return err
 			}
+		case writerTypeLegacyKV:
+			rp, err = newKVRowProcessor(ctx, flowCtx.Cfg, flowCtx.EvalCtx, lrw.spec, lrw.configByTable)
+			if err != nil {
+				return err
+			}
+		default:
+			return errors.AssertionFailedf("unknown logical replication writer type: %s", writer)
 		}
 
 		if streamingKnobs, ok := flowCtx.TestingKnobs().StreamingTestingKnobs.(*sql.StreamingTestingKnobs); ok {
@@ -760,6 +758,24 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 	return nil
 }
 
+func getWriterType(
+	ctx context.Context, mode jobspb.LogicalReplicationDetails_ApplyMode, settings *cluster.Settings,
+) (writerType, error) {
+	switch mode {
+	case jobspb.LogicalReplicationDetails_Immediate:
+		// Require v25.2 to use the sql writer by default to ensure that the
+		// KV origin timestamp validation is available on all nodes.
+		if settings.Version.IsActive(ctx, clusterversion.V25_2) {
+			return writerType(immediateModeWriter.Get(&settings.SV)), nil
+		}
+		return writerTypeSQL, nil
+	case jobspb.LogicalReplicationDetails_Validated:
+		return writerTypeSQL, nil
+	default:
+		return "", errors.Newf("unknown logical replication writer type: %s", mode)
+	}
+}
+
 // flushBuffer processes some or all of the events in the passed buffer, and
 // zeros out each event in the passed buffer for which it successfully completed
 // processing either by applying it or by sending it to a DLQ. If mustProcess is
@@ -987,7 +1003,7 @@ func (t replicationMutationType) String() string {
 func (lrw *logicalReplicationWriterProcessor) flushChunk(
 	ctx context.Context, bh BatchHandler, chunk []streampb.StreamEvent_KV, canRetry retryEligibility,
 ) (flushStats, error) {
-	batchSize := lrw.getBatchSize()
+	batchSize := bh.BatchSize()
 
 	lrw.debug.RecordChunkStart()
 	defer lrw.debug.RecordChunkComplete()
@@ -1213,20 +1229,14 @@ type BatchHandler interface {
 	// or are not applied as a group. If the batch is a single KV it may use an
 	// implicit txn.
 	HandleBatch(context.Context, []streampb.StreamEvent_KV) (batchStats, error)
+	BatchSize() int
 	GetLastRow() cdcevent.Row
 	SetSyntheticFailurePercent(uint32)
 	ReportMutations(*stats.Refresher)
 	ReleaseLeases(context.Context)
 	Close(context.Context)
 }
 
-var useImplicitTxns = settings.RegisterBoolSetting(
-	settings.ApplicationLevel,
-	"logical_replication.consumer.use_implicit_txns.enabled",
-	"determines whether the consumer processes each row in a separate implicit txn",
-	metamorphic.ConstantWithTestBool("logical_replication.consumer.use_implicit_txns.enabled", true),
-)
-
 func init() {
 	rowexec.NewLogicalReplicationWriterProcessor = newLogicalReplicationWriterProcessor
 }

pkg/crosscluster/logical/lww_kv_processor.go

@@ -133,6 +133,10 @@ func (p *kvRowProcessor) HandleBatch(
 	return batchStats{}, errors.AssertionFailedf("TODO: multi-row transactions not supported by the kvRowProcessor")
 }
 
+func (p *kvRowProcessor) BatchSize() int {
+	return 1
+}
+
 func (p *kvRowProcessor) processRow(
 	ctx context.Context,
 	txn isql.Txn,

pkg/crosscluster/logical/lww_row_processor.go

@@ -280,6 +280,10 @@ func (srp *sqlRowProcessor) ReleaseLeases(ctx context.Context) {
 	srp.querier.ReleaseLeases(ctx)
 }
 
+func (srp *sqlRowProcessor) BatchSize() int {
+	return int(flushBatchSize.Get(&srp.settings.SV))
+}
+
 func (*sqlRowProcessor) Close(ctx context.Context) {}
 
 var errInjected = errors.New("injected synthetic error")

pkg/settings/registry.go

@@ -258,13 +258,14 @@ var retiredSettings = map[InternalKey]struct{}{
 	"sql.metrics.statement_details.plan_collection.period":       {},
 
 	// removed as of 25.2
-	"kv.snapshot_receiver.excise.enabled":                {},
-	"sql.catalog.experimental_use_session_based_leasing": {},
-	"bulkio.backup.merge_file_buffer_size":               {},
-	"changefeed.new_webhook_sink_enabled":                {},
-	"changefeed.new_webhook_sink.enabled":                {},
-	"changefeed.new_pubsub_sink_enabled":                 {},
-	"changefeed.new_pubsub_sink.enabled":                 {},
+	"kv.snapshot_receiver.excise.enabled":                    {},
+	"sql.catalog.experimental_use_session_based_leasing":     {},
+	"bulkio.backup.merge_file_buffer_size":                   {},
+	"changefeed.new_webhook_sink_enabled":                    {},
+	"changefeed.new_webhook_sink.enabled":                    {},
+	"changefeed.new_pubsub_sink_enabled":                     {},
+	"changefeed.new_pubsub_sink.enabled":                     {},
+	"logical_replication.consumer.use_implicit_txns.enabled": {},
 }
 
 // grandfatheredDefaultSettings is the list of "grandfathered" existing sql.defaults

pkg/sql/logictest/testdata/logic_test/jsonb_path_query

@@ -1074,3 +1074,18 @@ query T
 SELECT jsonb_path_query('[{"a": 1, "b": 2}, {"a": 1}, {"b": 2}, {}]', '$[*] ? (exists(@.a) && exists(@.b))');
 ----
 {"a": 1, "b": 2}
+
+query T
+SELECT jsonb_path_query('{}', '(1 + 2 == 3) is unknown');
+----
+false
+
+query T
+SELECT jsonb_path_query('{}', '($ < 1) is unknown');
+----
+true
+
+query T
+SELECT jsonb_path_query('{}', '(null like_regex "^he.*$") is unknown');
+----
+true