Merge cockroachdb#51659 cockroachdb#51662 cockroachdb#51681 cockroach…

…db#51789

51659: colexec: fix closing of parallel unordered sync inputs r=jordanlewis a=yuzefovich

The responsibility of closing all `Closer`s within the synchronizer
input's tree lies on the goroutine that's handling that input.
Previously, we were mistakenly calling `Close` after we have notified
the wait groups that the handler-goroutine is done. This can result in
races between vectorized flow cleanup and cleaning up those closers.
This is now fixed.

Fixes: cockroachdb#51470.

Release note: None

51662: builtins: add a builtin to convert the hlc.Timestamp decimal to timestamp r=rohany a=rohany

Fixes cockroachdb#51317.

Release note (sql change): Add the builtin
`crdb_internal.approximate_timestamp` to
convert the decimal returned from the `crdb_internal_mvcc_timestamp`
system column into a `timestamp`.

51681: opt: disable zigzag joins when system columns are requested r=yuzefovich a=rohany

Fixes cockroachdb#51664.

The zigzagjoiner isn't currently equipped to produce system columns, so
disable generation of zigzagjoin memo entries if any system columns are
requested.

Release note: None

51789: sql: fix panic when evaluating check constraint during insert fast path r=mgartner a=RichardJCai

Fix panic due to out of bounds error when intercepting error message in
insert_fast_path.go.

No release note since the panic is not in prod.

Release note: None

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

docs/generated/sql/functions.md

@@ -1975,6 +1975,8 @@ developers and its definition may change without prior notice.</p>
 <p>Note that uses of this function disable server-side optimizations and
 may increase either contention or retry errors, or both.</p>
 </span></td></tr>
+<tr><td><a name="crdb_internal.approximate_timestamp"></a><code>crdb_internal.approximate_timestamp(timestamp: <a href="decimal.html">decimal</a>) &rarr; <a href="timestamp.html">timestamp</a></code></td><td><span class="funcdesc"><p>Converts the crdb_internal_mvcc_timestamp column into an approximate timestamp.</p>
+</span></td></tr>
 <tr><td><a name="crdb_internal.check_consistency"></a><code>crdb_internal.check_consistency(stats_only: <a href="bool.html">bool</a>, start_key: <a href="bytes.html">bytes</a>, end_key: <a href="bytes.html">bytes</a>) &rarr; tuple{int AS range_id, bytes AS start_key, string AS start_key_pretty, string AS status, string AS detail}</code></td><td><span class="funcdesc"><p>Runs a consistency check on ranges touching the specified key range. an empty start or end key is treated as the minimum and maximum possible, respectively. stats_only should only be set to false when targeting a small number of ranges to avoid overloading the cluster. Each returned row contains the range ID, the status (a roachpb.CheckConsistencyResponse_Status), and verbose detail.</p>
 <p>Example usage:
 SELECT * FROM crdb_internal.check_consistency(true, ‘\x02’, ‘\x04’)</p>

pkg/sql/colexec/parallel_unordered_synchronizer.go

@@ -215,9 +215,11 @@ func (s *ParallelUnorderedSynchronizer) init(ctx context.Context) {
 				if int(atomic.AddUint32(&s.numFinishedInputs, 1)) == len(s.inputs) {
 					close(s.batchCh)
 				}
+				// We need to close all of the closers of this input before we
+				// notify the wait groups.
+				input.ToClose.CloseAndLogOnErr(ctx, "parallel unordered synchronizer input")
 				s.internalWaitGroup.Done()
 				s.externalWaitGroup.Done()
-				input.ToClose.CloseAndLogOnErr(ctx, "parallel unordered synchronizer input")
 			}()
 			sendErr := func(err error) {
 				if strings.Contains(err.Error(), context.Canceled.Error()) && atomic.LoadInt32(&internalCancellation) == 1 {

pkg/sql/insert_fast_path.go

@@ -355,6 +355,12 @@ func (n *insertFastPathNode) enableAutoCommit() {
 func interceptAlterColumnTypeParseError(
 	insertCols []sqlbase.ColumnDescriptor, colNum int, err error,
 ) error {
+	// Only intercept the error if the column being inserted into
+	// is an actual column. This is to avoid checking on values that don't
+	// correspond to an actual column, for example a check constraint.
+	if colNum >= len(insertCols) {
+		return err
+	}
 	var insertCol sqlbase.ColumnDescriptor
 
 	// wrapParseError is a helper function that checks if an insertCol has the

pkg/sql/logictest/testdata/logic_test/check_constraints

@@ -337,3 +337,11 @@ CREATE TABLE t46675isnotnull (k int, a int, CHECK ((k, a) IS NOT NULL))
 # value.
 statement error pgcode 23514 pq: failed to satisfy CHECK constraint \(\(k, a\) IS NOT NULL\)
 INSERT INTO t46675isnotnull VALUES (1, NULL)
+
+# Regression test for #51690. Make sure we don't panic when a check constraint
+# errors during an insert using the fast path.
+statement ok
+CREATE TABLE t51690(x INT, y INT, CHECK(x / y = 1));
+
+statement error pq: division by zero
+INSERT INTO t51690 VALUES (1, 0)

pkg/sql/logictest/testdata/logic_test/mvcc

@@ -95,6 +95,11 @@ SELECT x, y, crdb_internal_mvcc_timestamp IS NOT NULL AS foo FROM t ORDER BY foo
 ----
 1 2 true
 
+query B
+SELECT crdb_internal.approximate_timestamp(crdb_internal_mvcc_timestamp) < now() FROM t
+----
+true
+
 # Ensure that standard lookup joins can produce the timestamp column.
 statement ok
 CREATE TABLE t2 (x INT, INDEX (x));

pkg/sql/opt/memo/check_expr.go

@@ -202,6 +202,18 @@ func (m *Memo) CheckExpr(e opt.Expr) {
 		if len(t.LeftEqCols) != len(t.RightEqCols) {
 			panic(errors.AssertionFailedf("zigzag join with mismatching eq columns"))
 		}
+		meta := m.Metadata()
+		left, right := meta.Table(t.LeftTable), meta.Table(t.RightTable)
+		for i := 0; i < left.ColumnCount(); i++ {
+			if cat.IsSystemColumn(left, i) && t.Cols.Contains(t.LeftTable.ColumnID(i)) {
+				panic(errors.AssertionFailedf("zigzag join should not contain system column"))
+			}
+		}
+		for i := 0; i < right.ColumnCount(); i++ {
+			if cat.IsSystemColumn(right, i) && t.Cols.Contains(t.RightTable.ColumnID(i)) {
+				panic(errors.AssertionFailedf("zigzag join should not contain system column"))
+			}
+		}
 
 	case *AggDistinctExpr:
 		if t.Input.Op() == opt.AggFilterOp {

pkg/sql/opt/xform/custom_funcs.go

@@ -2181,6 +2181,18 @@ func (c *CustomFuncs) GenerateZigzagJoins(
 		return
 	}
 
+	// Zigzag joins aren't currently equipped to produce system columns, so
+	// don't generate any if some system columns are requested.
+	foundSystemCol := false
+	scanPrivate.Cols.ForEach(func(colID opt.ColumnID) {
+		if cat.IsSystemColumn(tab, scanPrivate.Table.ColumnOrdinal(colID)) {
+			foundSystemCol = true
+		}
+	})
+	if foundSystemCol {
+		return
+	}
+
 	// Iterate through indexes, looking for those prefixed with fixedEq cols.
 	// Efficiently finding a set of indexes that make the most efficient zigzag
 	// join, with no limit on the number of indexes selected, is an instance of

pkg/sql/sem/builtins/builtins.go

@@ -3188,6 +3188,19 @@ may increase either contention or retry errors, or both.`,
 		},
 	),
 
+	"crdb_internal.approximate_timestamp": makeBuiltin(
+		tree.FunctionProperties{Category: categorySystemInfo},
+		tree.Overload{
+			Types:      tree.ArgTypes{{"timestamp", types.Decimal}},
+			ReturnType: tree.FixedReturnType(types.Timestamp),
+			Fn: func(ctx *tree.EvalContext, args tree.Datums) (tree.Datum, error) {
+				return tree.DecimalToInexactDTimestamp(args[0].(*tree.DDecimal))
+			},
+			Info:       "Converts the crdb_internal_mvcc_timestamp column into an approximate timestamp.",
+			Volatility: tree.VolatilityImmutable,
+		},
+	),
+
 	"crdb_internal.cluster_id": makeBuiltin(
 		tree.FunctionProperties{Category: categorySystemInfo},
 		tree.Overload{

pkg/sql/sem/tree/eval.go

@@ -3257,6 +3257,21 @@ func TimestampToDecimal(ts hlc.Timestamp) apd.Decimal {
 	return res
 }
 
+// DecimalToInexactDTimestamp is the inverse of TimestampToDecimal. It converts
+// a decimal constructed from an hlc.Timestamp into an approximate DTimestamp
+// containing the walltime of the hlc.Timestamp.
+func DecimalToInexactDTimestamp(d *DDecimal) (*DTimestamp, error) {
+	var coef big.Int
+	coef.Set(&d.Decimal.Coeff)
+	// The physical portion of the HLC is stored shifted up by 10^10, so shift
+	// it down and clear out the logical component.
+	coef.Div(&coef, big10E10)
+	if !coef.IsInt64() {
+		return nil, pgerror.Newf(pgcode.DatetimeFieldOverflow, "timestamp value out of range: %s", d.String())
+	}
+	return TimestampToInexactDTimestamp(hlc.Timestamp{WallTime: coef.Int64()}), nil
+}
+
 // TimestampToDecimalDatum is the same as TimestampToDecimal, but
 // returns a datum.
 func TimestampToDecimalDatum(ts hlc.Timestamp) *DDecimal {

pkg/sql/sem/tree/eval_test.go

@@ -17,6 +17,7 @@ import (
 	"regexp"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/opt/exec/execbuilder"
@@ -27,9 +28,13 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/datadriven"
+	"github.com/stretchr/testify/require"
 )
 
 func TestEval(t *testing.T) {
@@ -366,3 +371,19 @@ func TestEvalError(t *testing.T) {
 		}
 	}
 }
+
+func TestHLCTimestampDecimalRoundTrip(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	rng, _ := randutil.NewPseudoRand()
+	for i := 0; i < 100; i++ {
+		ts := hlc.Timestamp{WallTime: rng.Int63(), Logical: rng.Int31()}
+		dec := tree.TimestampToDecimalDatum(ts)
+		approx, err := tree.DecimalToInexactDTimestamp(dec)
+		require.NoError(t, err)
+		// The expected timestamp is at the microsecond precision.
+		expectedTsDatum := tree.MustMakeDTimestamp(timeutil.Unix(0, ts.WallTime), time.Microsecond)
+		require.True(t, expectedTsDatum.Equal(approx.Time))
+	}
+}