diff --git a/pkg/frontend/data_branch_hashdiff.go b/pkg/frontend/data_branch_hashdiff.go index 730cf60758fdc..16e1d65d744cf 100644 --- a/pkg/frontend/data_branch_hashdiff.go +++ b/pkg/frontend/data_branch_hashdiff.go @@ -51,6 +51,30 @@ type lcaProbeLayout struct { enumValues []string } +func sortDataBranchBatchByPrimaryKey(bat *batch.Batch, pkColIdx int, mp *mpool.MPool) error { + pkVec := bat.Vecs[pkColIdx] + if !isDataBranchFloatType(*pkVec.GetType()) { + return mergeutil.SortColumnsByIndex(bat.Vecs, pkColIdx, mp) + } + + identityVec := vector.NewVec(types.T_uint64.ToType()) + defer identityVec.Free(mp) + for row := range pkVec.Length() { + identity, isNull, err := dataBranchFloatPKIdentityAt(pkVec, row) + if err != nil { + return err + } + if err = vector.AppendFixed(identityVec, identity, isNull, mp); err != nil { + return err + } + } + + cols := make([]*vector.Vector, 0, len(bat.Vecs)+1) + cols = append(cols, bat.Vecs...) + cols = append(cols, identityVec) + return mergeutil.SortColumnsByIndex(cols, len(cols)-1, mp) +} + func (layout lcaProbeLayout) columnNameForTargetIndex(targetIdx int) (string, bool) { for i, candidateIdx := range layout.targetIdxes { if candidateIdx == targetIdx { @@ -289,8 +313,8 @@ func handleDelsOnLCA( valsBuf.WriteString(fmt.Sprintf("row(%d,", i)) for j := range tuple { - if err = formatValIntoString( - ses, tuple[j], colTypes[expandedPKColIdxes[j]], valsBuf, + if err = formatValIntoStringWithFloatCast( + ses, tuple[j], colTypes[expandedPKColIdxes[j]], valsBuf, true, ); err != nil { return nil, err } @@ -320,7 +344,7 @@ func handleDelsOnLCA( valsBuf.WriteString(fmt.Sprintf("row(%d,", i)) b := tBat.Vecs[0].GetRawBytesAt(i) val := types.DecodeValue(b, tBat.Vecs[0].GetType().Oid) - if err = formatValIntoString(ses, val, pkType, valsBuf); err != nil { + if err = formatValIntoStringWithFloatCast(ses, val, pkType, valsBuf, true); err != nil { return nil, err } valsBuf.WriteString(")") @@ -357,12 +381,14 @@ func handleDelsOnLCA( ) for i := range quotedPKNames { - sqlBuf.WriteString(fmt.Sprintf("lca.%s = ", quotedPKNames[i])) + left := fmt.Sprintf("lca.%s", quotedPKNames[i]) + right := fmt.Sprintf("pks.%s", quotedPKValueAliases[i]) if castType, ok := lcaProbeJoinCastType(colTypes[expandedPKColIdxes[i]]); ok { - sqlBuf.WriteString(fmt.Sprintf("cast(pks.%s as %s)", quotedPKValueAliases[i], castType)) - } else { - sqlBuf.WriteString(fmt.Sprintf("pks.%s", quotedPKValueAliases[i])) + right = fmt.Sprintf("cast(%s as %s)", right, castType) } + sqlBuf.WriteString(dataBranchSQLKeyEqual( + left, right, colTypes[expandedPKColIdxes[i]], + )) if i != len(quotedPKNames)-1 { sqlBuf.WriteString(" AND ") } @@ -1053,14 +1079,15 @@ func hashDiffIfHasLCA( wg sync.WaitGroup atomicErr atomic.Value - baseDeleteBatches []batchWithKind - baseUpdateBatches []batchWithKind + baseDeleteBatches []batchWithKind + baseUpdateBatches []batchWithKind + restoreMissingKeys = make(map[string]struct{}) ) handleBaseDeleteAndUpdates := func(wrapped batchWithKind) error { wrapped.side = diffSideBase - if err2 := mergeutil.SortColumnsByIndex( - wrapped.batch.Vecs, tblStuff.def.pkColIdx, ses.proc.Mp(), + if err2 := sortDataBranchBatchByPrimaryKey( + wrapped.batch, tblStuff.def.pkColIdx, ses.proc.Mp(), ); err2 != nil { return err2 } @@ -1076,6 +1103,50 @@ func hashDiffIfHasLCA( handleTarDeleteAndUpdates := func(wrapped batchWithKind) (err2 error) { wrapped.side = diffSideTarget var pickConflictBat *batch.Batch + if wrapped.kind == diffInsert && wrapped.fromUpdate && len(restoreMissingKeys) > 0 { + var keep []int64 + restoreBat := tblStuff.retPool.acquireRetBatch(tblStuff, false) + for rowIdx := range wrapped.batch.RowCount() { + key, keyErr := extractPKAsString(ses, tblStuff, wrapped.batch, rowIdx) + if keyErr != nil { + tblStuff.retPool.releaseRetBatch(restoreBat, false) + return keyErr + } + if _, restore := restoreMissingKeys[key]; !restore { + keep = append(keep, int64(rowIdx)) + continue + } + if err2 = restoreBat.UnionOne(wrapped.batch, int64(rowIdx), ses.proc.Mp()); err2 != nil { + tblStuff.retPool.releaseRetBatch(restoreBat, false) + return err2 + } + delete(restoreMissingKeys, key) + } + if restoreBat.Vecs[0].Length() > 0 { + restoreBat.SetRowCount(restoreBat.Vecs[0].Length()) + if stop, e := emitBatch(emit, batchWithKind{ + batch: restoreBat, + kind: diffInsert, + name: wrapped.name, + side: wrapped.side, + fromUpdate: true, + restoreMissing: true, + }, false, tblStuff.retPool); e != nil { + tblStuff.retPool.releaseRetBatch(wrapped.batch, false) + return e + } else if stop { + tblStuff.retPool.releaseRetBatch(wrapped.batch, false) + return nil + } + } else { + tblStuff.retPool.releaseRetBatch(restoreBat, false) + } + if len(keep) == 0 { + tblStuff.retPool.releaseRetBatch(wrapped.batch, false) + return nil + } + wrapped.batch.Shrink(keep, true) + } if len(baseUpdateBatches) == 0 && len(baseDeleteBatches) == 0 { // no need to check conflict if stop, e := emitBatch(emit, wrapped, false, tblStuff.retPool); e != nil { @@ -1086,8 +1157,8 @@ func hashDiffIfHasLCA( return nil } - if err2 = mergeutil.SortColumnsByIndex( - wrapped.batch.Vecs, tblStuff.def.pkColIdx, ses.proc.Mp(), + if err2 = sortDataBranchBatchByPrimaryKey( + wrapped.batch, tblStuff.def.pkColIdx, ses.proc.Mp(), ); err2 != nil { return err2 } @@ -1112,7 +1183,7 @@ func hashDiffIfHasLCA( i, j := 0, 0 for i < tarVec.Length() && j < baseVec.Length() { - if cmp, err3 = compareSingleValInVector( + if cmp, err3 = compareDataBranchPrimaryKeyInVectors( ctx, ses, i, j, tarVec, baseVec, ); err3 != nil { return @@ -1139,6 +1210,15 @@ func hashDiffIfHasLCA( i++ j++ } else if copt.conflictOpt.Opt == tree.CONFLICT_ACCEPT { + if tarWrapped.kind == diffDelete && tarWrapped.fromUpdate && + baseWrapped.kind == diffDelete && !baseWrapped.fromUpdate { + key, keyErr := extractPKAsString(ses, tblStuff, tarWrapped.batch, i) + if keyErr != nil { + err3 = keyErr + return + } + restoreMissingKeys[key] = struct{}{} + } if tarWrapped.kind == diffDelete && baseWrapped.kind == diffDelete && !tarWrapped.fromUpdate && !baseWrapped.fromUpdate { @@ -1254,11 +1334,6 @@ func hashDiffIfHasLCA( return false }) - if wrapped.batch.RowCount() == 0 { - tblStuff.retPool.releaseRetBatch(wrapped.batch, false) - return - } - if pickConflictBat != nil { if stop, e := emitBatch(emit, batchWithKind{ batch: pickConflictBat, @@ -1271,6 +1346,10 @@ func hashDiffIfHasLCA( return nil } } + if wrapped.batch.RowCount() == 0 { + tblStuff.retPool.releaseRetBatch(wrapped.batch, false) + return nil + } stop, e := emitBatch(emit, wrapped, false, tblStuff.retPool) if e != nil { @@ -1390,6 +1469,12 @@ func hashDiffIfHasLCA( if err = stepHandler(false); err != nil { return } + if len(restoreMissingKeys) != 0 { + return moerr.NewInternalErrorNoCtxf( + "data branch source update is missing %d replacement row(s)", + len(restoreMissingKeys), + ) + } // what can I do with these left base updates/inserts ? if copt.conflictOpt == nil { @@ -1721,10 +1806,11 @@ func findDeleteAndUpdateBat( return err2 } if err2 = send(batchWithKind{ - name: tblName, - side: side, - batch: updateBat, - kind: diffInsert, + name: tblName, + side: side, + batch: updateBat, + kind: diffInsert, + fromUpdate: tblStuff.def.pkKind != fakeKind, }); err2 != nil { return err2 } @@ -2092,6 +2178,7 @@ func diffDataHelper( tarBat *batch.Batch baseBat *batch.Batch baseDeleteBat *batch.Batch + tarUpdateBat *batch.Batch tarTuple types.Tuple baseTuple types.Tuple checkRet databranchutils.GetResult @@ -2099,6 +2186,11 @@ func diffDataHelper( tarBat = tblStuff.retPool.acquireRetBatch(tblStuff, false) baseBat = tblStuff.retPool.acquireRetBatch(tblStuff, false) + defer func() { + if tarUpdateBat != nil { + tblStuff.retPool.releaseRetBatch(tarUpdateBat, false) + } + }() if err2 = cursor.ForEach(func(key []byte, row []byte) error { select { @@ -2174,10 +2266,13 @@ func diffDataHelper( if baseDeleteBat == nil { baseDeleteBat = tblStuff.retPool.acquireRetBatch(tblStuff, false) } + if tarUpdateBat == nil { + tarUpdateBat = tblStuff.retPool.acquireRetBatch(tblStuff, false) + } if err2 = appendTupleToBat(ses, baseDeleteBat, baseTuple, tblStuff); err2 != nil { return err2 } - if err2 = appendTupleToBat(ses, tarBat, tarTuple, tblStuff); err2 != nil { + if err2 = appendTupleToBat(ses, tarUpdateBat, tarTuple, tblStuff); err2 != nil { return err2 } } else { @@ -2204,10 +2299,11 @@ func diffDataHelper( if baseDeleteBat != nil { if stop, err3 := emitBatch(emit, batchWithKind{ - batch: baseDeleteBat, - kind: diffDelete, - name: tblStuff.baseRel.GetTableName(), - side: diffSideBase, + batch: baseDeleteBat, + kind: diffDelete, + name: tblStuff.baseRel.GetTableName(), + side: diffSideBase, + fromUpdate: true, }, false, tblStuff.retPool); err3 != nil { return err3 } else if stop { @@ -2215,6 +2311,22 @@ func diffDataHelper( } } + if tarUpdateBat != nil { + stop, err3 := emitBatch(emit, batchWithKind{ + batch: tarUpdateBat, + kind: diffInsert, + name: tblStuff.tarRel.GetTableName(), + side: diffSideTarget, + fromUpdate: true, + }, false, tblStuff.retPool) + tarUpdateBat = nil + if err3 != nil { + return err3 + } else if stop { + return nil + } + } + if stop, err3 := emitBatch(emit, batchWithKind{ batch: tarBat, kind: diffInsert, diff --git a/pkg/frontend/data_branch_hashdiff_test.go b/pkg/frontend/data_branch_hashdiff_test.go index e59a5aaa4a3df..962e6b0cd76a3 100644 --- a/pkg/frontend/data_branch_hashdiff_test.go +++ b/pkg/frontend/data_branch_hashdiff_test.go @@ -17,6 +17,7 @@ package frontend import ( "context" "fmt" + "math" "strings" "sync" "sync/atomic" @@ -908,9 +909,10 @@ func (h *closeTrackingBranchHashmap) Close() error { } type capturedBatch struct { - kind string - side int - rows [][]any + kind string + side int + rows [][]any + fromUpdate bool } func TestRunLCAProbeWithReaderFallback_EarlyReturns(t *testing.T) { @@ -1514,6 +1516,48 @@ func TestHandleDelsOnLCA_SQLPaths(t *testing.T) { require.ErrorIs(t, err, wantErr) }) + t.Run("float primary key join matches NaN explicitly", func(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + tblStuff := newTestBranchTableStuff(ctrl) + tblStuff.lcaRel = mock_frontend.NewMockRelation(ctrl) + tblStuff.def.colTypes[0] = types.T_float64.ToType() + targetDef := tblStuff.tarRel.GetTableDef(context.Background()) + targetDef.Cols[0].Typ = plan.Type{Id: int32(types.T_float64)} + baseDef := tblStuff.baseRel.GetTableDef(context.Background()) + baseDef.Cols[0].Typ = plan.Type{Id: int32(types.T_float64)} + lcaDef := newTestBranchTableDef("lca_tbl", "name") + lcaDef.Cols[0].Typ = plan.Type{Id: int32(types.T_float64)} + tblStuff.lcaRel.(*mock_frontend.MockRelation).EXPECT().GetTableDef(gomock.Any()).Return(lcaDef).AnyTimes() + tblStuff.lcaRel.(*mock_frontend.MockRelation).EXPECT().GetTableID(gomock.Any()).Return(uint64(76)).AnyTimes() + + wantErr := moerr.NewInternalErrorNoCtx("stop after sql capture") + bh := mock_frontend.NewMockBackgroundExec(ctrl) + bh.EXPECT().Exec(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, sql string) error { + require.Contains(t, sql, + "values row(0,bit_cast(unhex('010000000000f87f') as double)),row(1,cast(1.25 as double))") + right := "cast(pks.`__mo_data_branch_pk_0` as DOUBLE)" + require.Contains(t, sql, dataBranchSQLKeyEqual("lca.`id`", right, types.T_float64.ToType())) + return wantErr + }). + Times(1) + + tBat := batch.NewWithSize(1) + tBat.Vecs[0] = vector.NewVec(types.T_float64.ToType()) + require.NoError(t, vector.AppendFixed(tBat.Vecs[0], math.NaN(), false, ses.proc.Mp())) + require.NoError(t, vector.AppendFixed(tBat.Vecs[0], 1.25, false, ses.proc.Mp())) + tBat.SetRowCount(2) + defer tBat.Clean(ses.proc.Mp()) + + _, err := handleDelsOnLCA( + context.Background(), ses, bh, tBat, tblStuff, + types.BuildTS(10, 0).ToTimestamp(), + ) + require.ErrorIs(t, err, wantErr) + }) + t.Run("internal aliases do not collide with user primary key", func(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -1984,7 +2028,9 @@ func TestHashDiff_NoLCABoundedUpdateKeepsLatestRow(t *testing.T) { rows := decodeCapturedRows(t, w.batch, tblStuff.def.colTypes) mu.Lock() if len(rows) > 0 { - got = append(got, capturedBatch{kind: w.kind, side: w.side, rows: rows}) + got = append(got, capturedBatch{ + kind: w.kind, side: w.side, rows: rows, fromUpdate: w.fromUpdate, + }) } mu.Unlock() tblStuff.retPool.releaseRetBatch(w.batch, false) @@ -2002,9 +2048,11 @@ func TestHashDiff_NoLCABoundedUpdateKeepsLatestRow(t *testing.T) { require.Len(t, got, 2) require.Equal(t, diffDelete, got[0].kind) require.Equal(t, diffSideBase, got[0].side) + require.True(t, got[0].fromUpdate) require.Equal(t, [][]any{{int64(1), "destination", "h1"}}, got[0].rows) require.Equal(t, diffInsert, got[1].kind) require.Equal(t, diffSideTarget, got[1].side) + require.True(t, got[1].fromUpdate) require.Equal(t, [][]any{{int64(1), "bounded", "h1"}}, got[1].rows) } diff --git a/pkg/frontend/data_branch_helpers.go b/pkg/frontend/data_branch_helpers.go index 8848d76330580..630346243a692 100644 --- a/pkg/frontend/data_branch_helpers.go +++ b/pkg/frontend/data_branch_helpers.go @@ -17,7 +17,11 @@ package frontend import ( "bytes" "context" + "encoding/binary" + "encoding/hex" "encoding/json" + "fmt" + "math" "reflect" "regexp" "strconv" @@ -41,6 +45,85 @@ import ( var snapConditionRegex = regexp.MustCompile(`\{[^}]+}`) +func isDataBranchFloatType(typ types.Type) bool { + return typ.Oid == types.T_float32 || typ.Oid == types.T_float64 +} + +// dataBranchSQLKeyEqual returns the SQL predicate for Data Branch key +// identity. MatrixOne primary keys preserve FLOAT/DOUBLE bits, so scalar +// equality is insufficient: it collapses signed zero and cannot distinguish +// NaN payloads. serial() is the same bit-preserving encoding used by key paths. +func dataBranchSQLKeyEqual(left, right string, typ types.Type) string { + if !isDataBranchFloatType(typ) { + return fmt.Sprintf("%s = %s", left, right) + } + return fmt.Sprintf("serial(%s) = serial(%s)", left, right) +} + +// dataBranchFloatPKIdentityAt returns the exact storage identity used to order +// a simple FLOAT/DOUBLE primary key during hash-diff conflict matching. Scalar +// float comparison is not a key comparison: it collapses signed zero and NaN +// payloads. The raw IEEE bits form a deterministic total order and preserve +// every legal stored key. +func dataBranchFloatPKIdentityAt(vec *vector.Vector, row int) (uint64, bool, error) { + if vec.IsConst() { + row = 0 + } + if vec.IsNull(uint64(row)) { + return 0, true, nil + } + switch vec.GetType().Oid { + case types.T_float32: + return uint64(math.Float32bits(vector.GetFixedAtNoTypeCheck[float32](vec, row))), false, nil + case types.T_float64: + return math.Float64bits(vector.GetFixedAtNoTypeCheck[float64](vec, row)), false, nil + default: + return 0, false, moerr.NewInternalErrorNoCtxf( + "data branch: exact float key identity requires FLOAT/DOUBLE, got %s", + vec.GetType().String(), + ) + } +} + +func compareDataBranchPrimaryKeyInVectors( + ctx context.Context, + ses *Session, + rowIdx1 int, + rowIdx2 int, + vec1 *vector.Vector, + vec2 *vector.Vector, +) (int, error) { + if !vec1.GetType().Eq(*vec2.GetType()) || !isDataBranchFloatType(*vec1.GetType()) { + return compareSingleValInVector(ctx, ses, rowIdx1, rowIdx2, vec1, vec2) + } + + left, leftNull, err := dataBranchFloatPKIdentityAt(vec1, rowIdx1) + if err != nil { + return 0, err + } + right, rightNull, err := dataBranchFloatPKIdentityAt(vec2, rowIdx2) + if err != nil { + return 0, err + } + if leftNull || rightNull { + switch { + case leftNull && rightNull: + return 0, nil + case leftNull: + return -1, nil + default: + return 1, nil + } + } + if left < right { + return -1, nil + } + if left > right { + return 1, nil + } + return 0, nil +} + func containsDataBranchTempTableName(sqlLower string) bool { return containsTempTableMarker(sqlLower, "__mo_diff_del_") || containsTempTableMarker(sqlLower, "__mo_diff_ins_") @@ -512,6 +595,20 @@ func scanSnapshotRelationByIDWithFallback( } func formatValIntoString(ses *Session, val any, t types.Type, buf *bytes.Buffer) error { + return formatValIntoStringWithFloatCast(ses, val, t, buf, false) +} + +// formatValIntoStringWithFloatCast can force finite FLOAT/DOUBLE values to +// carry an explicit SQL type. This is needed by VALUES probes: without a cast +// on every row, VALUES type inference can convert a NaN cell to its integer bit +// pattern before the caller casts the resulting column back to FLOAT/DOUBLE. +func formatValIntoStringWithFloatCast( + ses *Session, + val any, + t types.Type, + buf *bytes.Buffer, + castFiniteFloat bool, +) error { if val == nil { buf.WriteString("NULL") return nil @@ -527,8 +624,39 @@ func formatValIntoString(ses *Session, val any, t types.Type, buf *bytes.Buffer) buf.Write(strconv.AppendUint(scratch[:0], v, 10)) } - writeFloat := func(v float64, bitSize int) { + writeFloat := func(v float64, bits uint64, bitSize int, sqlType string) { + if math.IsNaN(v) || (v == 0 && math.Signbit(v)) { + var raw [8]byte + if bitSize == 32 { + binary.LittleEndian.PutUint32(raw[:4], uint32(bits)) + } else { + binary.LittleEndian.PutUint64(raw[:], bits) + } + buf.WriteString("bit_cast(unhex('") + buf.WriteString(hex.EncodeToString(raw[:bitSize/8])) + buf.WriteString("')") + buf.WriteString(" as ") + buf.WriteString(sqlType) + buf.WriteByte(')') + return + } + if math.IsInf(v, 0) { + buf.WriteString("cast('") + buf.Write(strconv.AppendFloat(scratch[:0], v, 'g', -1, bitSize)) + buf.WriteString("' as ") + buf.WriteString(sqlType) + buf.WriteByte(')') + return + } + if castFiniteFloat { + buf.WriteString("cast(") + } buf.Write(strconv.AppendFloat(scratch[:0], v, 'g', -1, bitSize)) + if castFiniteFloat { + buf.WriteString(" as ") + buf.WriteString(sqlType) + buf.WriteByte(')') + } } writeBool := func(v bool) { @@ -664,9 +792,11 @@ func formatValIntoString(ses *Session, val any, t types.Type, buf *bytes.Buffer) case types.T_int64: writeInt(val.(int64)) case types.T_float32: - writeFloat(float64(val.(float32)), 32) + v := val.(float32) + writeFloat(float64(v), uint64(math.Float32bits(v)), 32, "float") case types.T_float64: - writeFloat(val.(float64), 64) + v := val.(float64) + writeFloat(v, math.Float64bits(v), 64, "double") case types.T_array_float32: buf.WriteString("'") buf.WriteString(types.ArrayToString[float32](val.([]float32))) diff --git a/pkg/frontend/data_branch_helpers_test.go b/pkg/frontend/data_branch_helpers_test.go index 923d4d31d994f..3a697a18024df 100644 --- a/pkg/frontend/data_branch_helpers_test.go +++ b/pkg/frontend/data_branch_helpers_test.go @@ -56,6 +56,17 @@ func TestAcquireReleaseBuffer(t *testing.T) { }) } +func TestDataBranchSQLKeyEqual(t *testing.T) { + require.Equal(t, "left_key = right_key", + dataBranchSQLKeyEqual("left_key", "right_key", types.T_int64.ToType())) + + for _, typ := range []types.Type{types.T_float32.ToType(), types.T_float64.ToType()} { + require.Equal(t, "serial(left_key) = serial(right_key)", + dataBranchSQLKeyEqual("left_key", "right_key", typ), + ) + } +} + func TestNewEmitter(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() diff --git a/pkg/frontend/data_branch_output.go b/pkg/frontend/data_branch_output.go index c5157cc79efb1..00717fe5fb629 100644 --- a/pkg/frontend/data_branch_output.go +++ b/pkg/frontend/data_branch_output.go @@ -138,14 +138,16 @@ func truncateDiffFileNamePrefix(name string, maxBytes int) string { } type applyBatchInfo struct { - dbName string - baseTable string - deleteTable string - insertTable string - deleteKeyNames []string - deleteStageNames []string - writableNames []string - disableInsertStage bool + dbName string + baseTable string + deleteTable string + insertTable string + deleteKeyNames []string + deleteStageNames []string + deleteKeyTypes []types.Type + writableNames []string + disableInsertStage bool + insertRowsIndividually bool } func newSQLValuesAppender( @@ -219,10 +221,20 @@ func newApplyBatchInfo( deleteKeyNames := make([]string, len(deleteKeyColIdxes)) deleteStageNames := make([]string, len(deleteKeyColIdxes)) + deleteKeyTypes := make([]types.Type, len(deleteKeyColIdxes)) + insertRowsIndividually := false for i, idx := range deleteKeyColIdxes { deleteKeyNames[i] = tblStuff.def.baseColNames[idx] deleteStageNames[i] = fmt.Sprintf("branch_apply_key_%d", i) + deleteKeyTypes[i] = tblStuff.def.colTypes[idx] + insertRowsIndividually = insertRowsIndividually || isDataBranchFloatType(deleteKeyTypes[i]) } + // MatrixOne accepts bit-distinct FLOAT/DOUBLE primary keys when each row is + // inserted independently. Multi-row INSERT and INSERT ... SELECT currently + // compare those keys with scalar float semantics, which collapses NaN + // payloads and signed zero. Keep the generated apply path aligned with the + // bit-preserving primary-key identity used by storage. + disableInsertStage = disableInsertStage || insertRowsIndividually writableIdxes := tblStuff.def.writableIdxes if len(tblStuff.def.tarOnlyIdxes) > 0 { @@ -236,25 +248,65 @@ func newApplyBatchInfo( seq := atomic.AddUint64(&diffTempTableSeq, 1) sessionTag := strings.ReplaceAll(ses.GetUUIDString(), "-", "") return &applyBatchInfo{ - dbName: tblStuff.baseRel.GetTableDef(ctx).DbName, - baseTable: tblStuff.baseRel.GetTableName(), - deleteTable: fmt.Sprintf("__mo_diff_del_%s_%d", sessionTag, seq), - insertTable: fmt.Sprintf("__mo_diff_ins_%s_%d", sessionTag, seq), - deleteKeyNames: deleteKeyNames, - deleteStageNames: deleteStageNames, - writableNames: writableNames, - disableInsertStage: disableInsertStage, + dbName: tblStuff.baseRel.GetTableDef(ctx).DbName, + baseTable: tblStuff.baseRel.GetTableName(), + deleteTable: fmt.Sprintf("__mo_diff_del_%s_%d", sessionTag, seq), + insertTable: fmt.Sprintf("__mo_diff_ins_%s_%d", sessionTag, seq), + deleteKeyNames: deleteKeyNames, + deleteStageNames: deleteStageNames, + deleteKeyTypes: deleteKeyTypes, + writableNames: writableNames, + disableInsertStage: disableInsertStage, + insertRowsIndividually: insertRowsIndividually, } } -func (batchInfo *applyBatchInfo) effectiveDeleteStageNames() []string { - if batchInfo == nil { - return nil +func (batchInfo *applyBatchInfo) validateDeleteKeyLayout() error { + if batchInfo == nil || len(batchInfo.deleteKeyNames) == 0 || + len(batchInfo.deleteStageNames) != len(batchInfo.deleteKeyNames) || + len(batchInfo.deleteKeyTypes) != len(batchInfo.deleteKeyNames) { + return moerr.NewInternalErrorNoCtx("invalid Data Branch staged delete key layout") + } + return nil +} + +func (batchInfo *applyBatchInfo) deleteNeedsExactFloatKeyMatch() bool { + for _, typ := range batchInfo.deleteKeyTypes { + if isDataBranchFloatType(typ) { + return true + } } - if len(batchInfo.deleteStageNames) == len(batchInfo.deleteKeyNames) && len(batchInfo.deleteStageNames) > 0 { - return batchInfo.deleteStageNames + return false +} + +func (batchInfo *applyBatchInfo) stagedDeleteSQL(baseTable, deleteTable string) (string, error) { + if err := batchInfo.validateDeleteKeyLayout(); err != nil { + return "", err + } + deleteStageNames := batchInfo.deleteStageNames + if !batchInfo.deleteNeedsExactFloatKeyMatch() { + pkExpr := quoteIdentifierForSQL(batchInfo.deleteKeyNames[0]) + if len(batchInfo.deleteKeyNames) > 1 { + pkExpr = fmt.Sprintf("(%s)", joinQuotedColumnNames(batchInfo.deleteKeyNames)) + } + return fmt.Sprintf( + "delete from %s where %s in (select %s from %s)", + baseTable, pkExpr, joinQuotedColumnNames(deleteStageNames), deleteTable, + ), nil + } + + const baseAlias = "branch_apply_base" + const stageAlias = "branch_apply_stage" + predicates := make([]string, len(batchInfo.deleteKeyNames)) + for i := range batchInfo.deleteKeyNames { + left := fmt.Sprintf("%s.%s", baseAlias, quoteIdentifierForSQL(batchInfo.deleteKeyNames[i])) + right := fmt.Sprintf("%s.%s", stageAlias, quoteIdentifierForSQL(deleteStageNames[i])) + predicates[i] = dataBranchSQLKeyEqual(left, right, batchInfo.deleteKeyTypes[i]) } - return batchInfo.deleteKeyNames + return fmt.Sprintf( + "delete %s from %s as %s join %s as %s on %s", + baseAlias, baseTable, baseAlias, deleteTable, stageAlias, strings.Join(predicates, " AND "), + ), nil } func mergeDiffs( @@ -1734,6 +1786,7 @@ func writeDeleteRowSQLFull( tblStuff.baseRel.GetTableDef(ctx).Name, ), )) + var literal bytes.Buffer for i, idx := range tblStuff.def.visibleIdxes { if i > 0 { buf.WriteString(" and ") @@ -1743,11 +1796,13 @@ func writeDeleteRowSQLFull( buf.WriteString(colName) buf.WriteString(" is null") } else { - buf.WriteString(colName) - buf.WriteString(" = ") - if err := formatValIntoString(ses, row[idx], tblStuff.def.colTypes[idx], buf); err != nil { + literal.Reset() + if err := formatValIntoString(ses, row[idx], tblStuff.def.colTypes[idx], &literal); err != nil { return err } + buf.WriteString(dataBranchSQLKeyEqual( + colName, literal.String(), tblStuff.def.colTypes[idx], + )) } } buf.WriteString(" limit 1;\n") @@ -1863,6 +1918,15 @@ func appendBatchRowsAsSQLValues( tmpValsBuffer *bytes.Buffer, appender sqlValuesAppender, ) (err error) { + exactFloatKeyUpdate, err := dataBranchExactFloatKeyUpdateBatch(wrapped, appender.batchInfo) + if err != nil { + return err + } + if exactFloatKeyUpdate { + if wrapped.kind == diffDelete { + return nil + } + } //seenCols := make(map[int]struct{}, len(tblStuff.def.visibleIdxes)) row := make([]any, len(tblStuff.def.colNames)) @@ -1878,8 +1942,9 @@ func appendBatchRowsAsSQLValues( ); err != nil { return } - if err = appendDataBranchApplyRowAsSQLValues( + if err = appendOrExecuteDataBranchApplyRow( ctx, ses, tblStuff, wrapped.kind, row, tmpValsBuffer, appender, + exactFloatKeyUpdate, wrapped.restoreMissing, ); err != nil { return } @@ -1888,6 +1953,135 @@ func appendBatchRowsAsSQLValues( return nil } +func dataBranchExactFloatKeyUpdateBatch( + wrapped batchWithKind, + batchInfo *applyBatchInfo, +) (bool, error) { + if !wrapped.fromUpdate || batchInfo == nil || !batchInfo.deleteNeedsExactFloatKeyMatch() { + return false, nil + } + if wrapped.kind != diffDelete && wrapped.kind != diffInsert { + return false, moerr.NewInternalErrorNoCtxf("unexpected Data Branch update batch kind %q", wrapped.kind) + } + return true, nil +} + +func appendOrExecuteDataBranchApplyRow( + ctx context.Context, + ses *Session, + tblStuff tableStuff, + kind string, + row []any, + tmpValsBuffer *bytes.Buffer, + appender sqlValuesAppender, + exactFloatKeyUpdate bool, + restoreMissing bool, +) error { + if !exactFloatKeyUpdate { + return appendDataBranchApplyRowAsSQLValues( + ctx, ses, tblStuff, kind, row, tmpValsBuffer, appender, + ) + } + + statements, err := exactFloatKeyUpdateSQL( + ctx, ses, tblStuff, row, tmpValsBuffer, restoreMissing, + ) + if err != nil { + return err + } + return execSQLStatements(ctx, ses, appender.bh, appender.writeFile, statements) +} + +// exactFloatKeyUpdateSQL applies a source update without passing FLOAT/DOUBLE +// identity through scalar equality. A row marked restoreMissing is known by the +// diff to have been independently deleted from the destination and is restored +// with one direct INSERT ... VALUES. The primary-key constraint plan compares +// FLOAT/DOUBLE serial encodings, so this path retains bit-distinct peers. +func exactFloatKeyUpdateSQL( + ctx context.Context, + ses *Session, + tblStuff tableStuff, + row []any, + buf *bytes.Buffer, + restoreMissing bool, +) ([]string, error) { + writableIdxes := tblStuff.def.writableIdxes + if len(tblStuff.def.tarOnlyIdxes) > 0 { + writableIdxes = tblStuff.def.commonWritableIdxes + } + qualifiedName := qualifiedTableName( + tblStuff.baseRel.GetTableDef(ctx).DbName, + tblStuff.baseRel.GetTableDef(ctx).Name, + ) + + buf.Reset() + buf.WriteString("update ") + buf.WriteString(qualifiedName) + buf.WriteString(" set ") + written := 0 + for _, idx := range writableIdxes { + if slices.Contains(tblStuff.def.pkColIdxes, idx) { + continue + } + if written > 0 { + buf.WriteString(",") + } + buf.WriteString(quoteIdentifierForSQL(tblStuff.def.baseColNames[idx])) + buf.WriteString(" = ") + if err := formatValIntoString(ses, row[idx], tblStuff.def.colTypes[idx], buf); err != nil { + return nil, err + } + written++ + } + if written == 0 { + return nil, moerr.NewInternalErrorNoCtx("Data Branch update has no writable non-key columns") + } + buf.WriteString(" where ") + if err := writeExactDataBranchKeyPredicate(ses, tblStuff, row, buf); err != nil { + return nil, err + } + buf.WriteString(" limit 1") + updateSQL := buf.String() + if !restoreMissing { + return []string{updateSQL}, nil + } + + buf.Reset() + buf.WriteString("insert into ") + buf.WriteString(qualifiedName) + buf.WriteString(" (") + buf.WriteString(strings.Join(quotedBaseColumnNamesByIdxes(tblStuff, writableIdxes), ",")) + buf.WriteString(") values ") + if err := writeInsertRowValues(ses, tblStuff, row, buf, writableIdxes); err != nil { + return nil, err + } + return []string{buf.String()}, nil +} + +func writeExactDataBranchKeyPredicate( + ses *Session, + tblStuff tableStuff, + row []any, + buf *bytes.Buffer, +) error { + var literal bytes.Buffer + for i, idx := range tblStuff.def.pkColIdxes { + if i > 0 { + buf.WriteString(" and ") + } + literal.Reset() + if err := formatValIntoString(ses, row[idx], tblStuff.def.colTypes[idx], &literal); err != nil { + return err + } + buf.WriteString(dataBranchSQLKeyEqual( + quoteIdentifierForSQL(tblStuff.def.baseColNames[idx]), + literal.String(), + tblStuff.def.colTypes[idx], + )) + } + return nil +} + func prepareFSForDiffAsFile( ctx context.Context, ses *Session, @@ -2148,8 +2342,10 @@ func tryFlushDeletesOrInserts( return flushDeletes() } } else { + insertRowsIndividually := batchInfo != nil && + batchInfo.insertRowsIndividually && *insertCnt > 0 if insertBuf.Len()+newValsLen >= maxSqlBatchSize || - *insertCnt+newRowCnt >= maxSqlBatchCnt { + *insertCnt+newRowCnt >= maxSqlBatchCnt || insertRowsIndividually { if *deleteCnt > 0 { if err = flushDeletes(); err != nil { return err @@ -2206,6 +2402,20 @@ func writeInsertRowValues( idxes []int, ) error { buf.WriteString("(") + if err := writeRowValueList(ses, tblStuff, row, buf, idxes); err != nil { + return err + } + buf.WriteString(")") + return nil +} + +func writeRowValueList( + ses *Session, + tblStuff tableStuff, + row []any, + buf *bytes.Buffer, + idxes []int, +) error { for i, idx := range idxes { if err := formatValIntoString(ses, row[idx], tblStuff.def.colTypes[idx], buf); err != nil { return err @@ -2214,8 +2424,6 @@ func writeInsertRowValues( buf.WriteString(",") } } - buf.WriteString(")") - return nil } @@ -2374,12 +2582,15 @@ func initApplyTables( if batchInfo == nil { return nil } + if err := batchInfo.validateDeleteKeyLayout(); err != nil { + return err + } baseTable := qualifiedTableName(batchInfo.dbName, batchInfo.baseTable) deleteTable := qualifiedTableName(batchInfo.dbName, batchInfo.deleteTable) insertTable := qualifiedTableName(batchInfo.dbName, batchInfo.insertTable) - deleteStageNames := batchInfo.effectiveDeleteStageNames() + deleteStageNames := batchInfo.deleteStageNames deleteSelectExprs := make([]string, len(batchInfo.deleteKeyNames)) for i := range batchInfo.deleteKeyNames { deleteSelectExprs[i] = fmt.Sprintf( @@ -2471,18 +2682,13 @@ func flushSqlValues( baseTable := qualifiedTableName(batchInfo.dbName, batchInfo.baseTable) deleteTable := qualifiedTableName(batchInfo.dbName, batchInfo.deleteTable) insertTable := qualifiedTableName(batchInfo.dbName, batchInfo.insertTable) - deleteStageNames := batchInfo.effectiveDeleteStageNames() if isDeleteFrom { insertStmt := fmt.Sprintf("insert into %s values %s", deleteTable, buf.String()) - pkExpr := quoteIdentifierForSQL(batchInfo.deleteKeyNames[0]) - if len(batchInfo.deleteKeyNames) > 1 { - pkExpr = fmt.Sprintf("(%s)", joinQuotedColumnNames(batchInfo.deleteKeyNames)) + deleteStmt, err := batchInfo.stagedDeleteSQL(baseTable, deleteTable) + if err != nil { + return err } - deleteStmt := fmt.Sprintf( - "delete from %s where %s in (select %s from %s)", - baseTable, pkExpr, joinQuotedColumnNames(deleteStageNames), deleteTable, - ) clearStmt := fmt.Sprintf("delete from %s", deleteTable) return execSQLStatements(ctx, ses, bh, writeFile, []string{insertStmt, deleteStmt, clearStmt}) } diff --git a/pkg/frontend/data_branch_output_test.go b/pkg/frontend/data_branch_output_test.go index d440415c2de41..0e39cfcae7564 100644 --- a/pkg/frontend/data_branch_output_test.go +++ b/pkg/frontend/data_branch_output_test.go @@ -19,6 +19,7 @@ import ( "context" "errors" "io" + "math" "os" "path/filepath" "regexp" @@ -1351,6 +1352,34 @@ func TestDataBranchOutputWriteDeleteRowSQLFull(t *testing.T) { buf := &bytes.Buffer{} require.NoError(t, writeDeleteRowSQLFull(context.Background(), nil, tblStuff, row, buf)) require.Equal(t, "delete from `db1`.`t1` where `id` = 9 and `name` is null limit 1;\n", buf.String()) + + tblStuff.def.colNames = []string{"f32", "f64", "nullable", "pos_inf", "neg_inf"} + tblStuff.def.baseColNames = []string{"f32", "f64", "nullable", "pos_inf", "neg_inf"} + tblStuff.def.colTypes = []types.Type{ + types.T_float32.ToType(), + types.T_float64.ToType(), + types.T_float64.ToType(), + types.T_float32.ToType(), + types.T_float64.ToType(), + } + tblStuff.def.visibleIdxes = []int{0, 1, 2, 3, 4} + + row = []any{ + math.Float32frombits(0x7fc00001), + math.Float64frombits(0x7ff8000000000001), + nil, + float32(math.Inf(1)), + math.Inf(-1), + } + buf.Reset() + require.NoError(t, writeDeleteRowSQLFull(context.Background(), nil, tblStuff, row, buf)) + require.Equal(t, + "delete from `db1`.`t1` where serial(`f32`) = serial(bit_cast(unhex('0100c07f') as float)) and "+ + "serial(`f64`) = serial(bit_cast(unhex('010000000000f87f') as double)) and `nullable` is null and "+ + "serial(`pos_inf`) = serial(cast('+Inf' as float)) and "+ + "serial(`neg_inf`) = serial(cast('-Inf' as double)) limit 1;\n", + buf.String(), + ) } func TestDataBranchOutputExecSQLStatementsWithWriteFile(t *testing.T) { @@ -1371,6 +1400,59 @@ func TestDataBranchOutputExecSQLStatementsWithWriteFile(t *testing.T) { require.Equal(t, "select 1;\ninsert into t values (1);\n", out.String()) } +func TestDataBranchOutputExactFloatKeyUpdateSQL(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + baseRel := mock_frontend.NewMockRelation(ctrl) + baseRel.EXPECT().GetTableDef(gomock.Any()).Return(&plan.TableDef{ + DbName: "db1", + Name: "t1", + }).AnyTimes() + + tblStuff := tableStuff{baseRel: baseRel} + tblStuff.def.baseColNames = []string{"f32", "f64", "tag", "note"} + tblStuff.def.colTypes = []types.Type{ + types.T_float32.ToType(), + types.T_float64.ToType(), + types.T_int64.ToType(), + types.T_varchar.ToType(), + } + tblStuff.def.pkColIdxes = []int{0, 1, 2} + tblStuff.def.writableIdxes = []int{0, 1, 2, 3} + + row := []any{ + math.Float32frombits(0x7fc00001), + math.Float64frombits(0x8000000000000000), + int64(7), + "updated", + } + var buf bytes.Buffer + statements, err := exactFloatKeyUpdateSQL( + context.Background(), nil, tblStuff, row, &buf, true, + ) + require.NoError(t, err) + require.Equal(t, + []string{ + "insert into `db1`.`t1` (`f32`,`f64`,`tag`,`note`) values (" + + "bit_cast(unhex('0100c07f') as float)," + + "bit_cast(unhex('0000000000000080') as double),7,'updated')", + }, + statements, + ) + + statements, err = exactFloatKeyUpdateSQL( + context.Background(), nil, tblStuff, row, &buf, false, + ) + require.NoError(t, err) + require.Equal(t, []string{ + "update `db1`.`t1` set `note` = 'updated' where " + + "serial(`f32`) = serial(bit_cast(unhex('0100c07f') as float)) and " + + "serial(`f64`) = serial(bit_cast(unhex('0000000000000080') as double)) and " + + "`tag` = 7 limit 1", + }, statements) +} + func TestDataBranchOutputInitAndDropApplyTablesWithWriteFile(t *testing.T) { batchInfo := &applyBatchInfo{ dbName: "db1", @@ -1379,6 +1461,7 @@ func TestDataBranchOutputInitAndDropApplyTablesWithWriteFile(t *testing.T) { insertTable: "__mo_diff_ins_x", deleteKeyNames: []string{"id"}, deleteStageNames: []string{"branch_apply_key_0"}, + deleteKeyTypes: []types.Type{types.T_int64.ToType()}, writableNames: []string{"id", "name"}, } @@ -1425,6 +1508,7 @@ func TestDataBranchOutputFlushSqlValuesWithWriteFile(t *testing.T) { insertTable: "__mo_diff_ins_x", deleteKeyNames: []string{"id", "name"}, deleteStageNames: []string{"branch_apply_key_0", "branch_apply_key_1"}, + deleteKeyTypes: []types.Type{types.T_int64.ToType(), types.T_varchar.ToType()}, writableNames: []string{"id", "name"}, } @@ -1475,16 +1559,78 @@ func TestDataBranchOutputFlushSqlValuesWithWriteFile(t *testing.T) { require.Contains(t, got, "insert into `db1`.`__mo_diff_del_x` values (1,'a');\n") require.Contains(t, got, "delete from `db1`.`t1` where (`id`,`name`) in (select `branch_apply_key_0`,`branch_apply_key_1` from `db1`.`__mo_diff_del_x`);\n") require.Contains(t, got, "insert into `db1`.`t1` (`id`,`name`) values (2,'b');\n") + + out.Reset() + floatBatchInfo := *batchInfo + floatBatchInfo.disableInsertStage = true + floatBatchInfo.insertRowsIndividually = true + deleteCnt, insertCnt := 0, 0 + deleteBuf, insertBuf := &bytes.Buffer{}, &bytes.Buffer{} + appender := sqlValuesAppender{ + ctx: context.Background(), tblStuff: tblStuff, batchInfo: &floatBatchInfo, + deleteCnt: &deleteCnt, deleteBuf: deleteBuf, insertCnt: &insertCnt, + insertBuf: insertBuf, writeFile: writeFile, + } + require.NoError(t, appender.appendRow(diffInsert, []byte("(1,'first')"))) + require.NoError(t, appender.appendRow(diffInsert, []byte("(2,'second')"))) + require.NoError(t, appender.flushAll()) + require.Equal(t, 2, strings.Count(out.String(), "insert into `db1`.`t1` (`id`,`name`) values ")) + require.NotContains(t, out.String(), "(1,'first'),(2,'second')") + require.NotContains(t, out.String(), "__mo_diff_ins_x") +} + +func TestDataBranchOutputFlushSqlValuesUsesExactFloatKeyMatch(t *testing.T) { + batchInfo := &applyBatchInfo{ + dbName: "db1", + baseTable: "t1", + deleteTable: "__mo_diff_del_x", + deleteKeyNames: []string{"float_key", "double_key", "int_key"}, + deleteStageNames: []string{"branch_apply_key_0", "branch_apply_key_1", "branch_apply_key_2"}, + deleteKeyTypes: []types.Type{ + types.T_float32.ToType(), + types.T_float64.ToType(), + types.T_int64.ToType(), + }, + } + + var out bytes.Buffer + require.NoError(t, flushSqlValues( + context.Background(), nil, nil, tableStuff{}, bytes.NewBufferString("(1,2,3)"), + true, false, batchInfo, func(b []byte) error { + _, err := out.Write(b) + return err + }, + )) + + got := out.String() + require.Contains(t, got, "insert into `db1`.`__mo_diff_del_x` values (1,2,3);\n") + require.Contains(t, got, + "delete branch_apply_base from `db1`.`t1` as branch_apply_base join `db1`.`__mo_diff_del_x` as branch_apply_stage on "+ + "serial(branch_apply_base.`float_key`) = serial(branch_apply_stage.`branch_apply_key_0`) AND "+ + "serial(branch_apply_base.`double_key`) = serial(branch_apply_stage.`branch_apply_key_1`) AND "+ + "branch_apply_base.`int_key` = branch_apply_stage.`branch_apply_key_2`;\n") + require.Contains(t, got, "delete from `db1`.`__mo_diff_del_x`;\n") +} + +func TestDataBranchOutputStagedDeleteRejectsIncompleteKeyLayout(t *testing.T) { + _, err := (&applyBatchInfo{ + deleteKeyNames: []string{"id"}, + deleteStageNames: []string{"branch_apply_key_0"}, + }).stagedDeleteSQL("`db1`.`t1`", "`db1`.`delete_stage`") + require.Error(t, err) + require.Contains(t, err.Error(), "invalid Data Branch staged delete key layout") } func TestDataBranchOutputTryFlushDeletesOrInserts(t *testing.T) { batchInfo := &applyBatchInfo{ - dbName: "db1", - baseTable: "t1", - deleteTable: "__mo_diff_del_x", - insertTable: "__mo_diff_ins_x", - deleteKeyNames: []string{"id"}, - writableNames: []string{"id", "name"}, + dbName: "db1", + baseTable: "t1", + deleteTable: "__mo_diff_del_x", + insertTable: "__mo_diff_ins_x", + deleteKeyNames: []string{"id"}, + deleteStageNames: []string{"branch_apply_key_0"}, + deleteKeyTypes: []types.Type{types.T_int64.ToType()}, + writableNames: []string{"id", "name"}, } t.Run("force flush both buffers", func(t *testing.T) { @@ -1520,7 +1666,7 @@ func TestDataBranchOutputTryFlushDeletesOrInserts(t *testing.T) { require.Equal(t, 0, insertCnt) require.Equal(t, 0, deleteBuf.Len()) require.Equal(t, 0, insertBuf.Len()) - require.Contains(t, out.String(), "delete from `db1`.`t1` where `id` in (select `id` from `db1`.`__mo_diff_del_x`);\n") + require.Contains(t, out.String(), "delete from `db1`.`t1` where `id` in (select `branch_apply_key_0` from `db1`.`__mo_diff_del_x`);\n") require.Contains(t, out.String(), "insert into `db1`.`t1` (`id`,`name`) select `id`,`name` from `db1`.`__mo_diff_ins_x`;\n") }) @@ -1611,6 +1757,11 @@ func TestDataBranchOutputBuildDataBranchApplyLayout(t *testing.T) { } tblStuff.def.colNames = []string{"id", "name", "age"} tblStuff.def.baseColNames = []string{"id", "name", "age"} + tblStuff.def.colTypes = []types.Type{ + types.T_float32.ToType(), + types.T_varchar.ToType(), + types.T_float64.ToType(), + } tblStuff.def.pkColIdxes = []int{0, 2} tblStuff.def.visibleIdxes = []int{0, 1, 2} tblStuff.def.writableIdxes = []int{0, 1, 2} @@ -1625,9 +1776,11 @@ func TestDataBranchOutputBuildDataBranchApplyLayout(t *testing.T) { require.Equal(t, "db1", info.dbName) require.Equal(t, "t1", info.baseTable) require.Equal(t, []string{"id", "age"}, info.deleteKeyNames) + require.Equal(t, []types.Type{types.T_float32.ToType(), types.T_float64.ToType()}, info.deleteKeyTypes) require.Equal(t, []string{"branch_apply_key_0", "branch_apply_key_1"}, info.deleteStageNames) require.Equal(t, []string{"id", "name", "age"}, info.writableNames) - require.False(t, info.disableInsertStage) + require.True(t, info.disableInsertStage) + require.True(t, info.insertRowsIndividually) require.True(t, strings.HasPrefix(info.deleteTable, "__mo_diff_del_")) require.True(t, strings.HasPrefix(info.insertTable, "__mo_diff_ins_")) @@ -1642,6 +1795,7 @@ func TestDataBranchOutputBuildDataBranchApplyLayout(t *testing.T) { require.Equal(t, []string{"branch_apply_key_0"}, info.deleteStageNames) require.Equal(t, []string{"id", "name"}, info.writableNames) require.True(t, info.disableInsertStage) + require.False(t, info.insertRowsIndividually) deleteByFullRow, deleteKeyColIdxes, info = buildDataBranchApplyLayout( context.Background(), &Session{}, fakeTblStuff, dataBranchApplyModeOnlinePKOnly, @@ -1650,6 +1804,7 @@ func TestDataBranchOutputBuildDataBranchApplyLayout(t *testing.T) { require.Equal(t, []int{0, 1}, deleteKeyColIdxes) require.NotNil(t, info) require.False(t, info.disableInsertStage) + require.False(t, info.insertRowsIndividually) deleteByFullRow, deleteKeyColIdxes, info = buildDataBranchApplyLayout( context.Background(), &Session{}, fakeTblStuff, dataBranchApplyModePortableSQL, @@ -1683,6 +1838,7 @@ func TestDataBranchApplyLayoutUsesDestinationColumnNames(t *testing.T) { tblStuff := tableStuff{baseRel: baseRel} tblStuff.def.colNames = []string{"id", tc.sourceName} tblStuff.def.baseColNames = []string{"id", tc.destinationName} + tblStuff.def.colTypes = []types.Type{types.T_int64.ToType(), types.T_varchar.ToType()} tblStuff.def.pkColIdxes = []int{0} tblStuff.def.visibleIdxes = []int{0, 1} tblStuff.def.writableIdxes = []int{0, 1} @@ -1700,12 +1856,14 @@ func TestDataBranchApplyLayoutUsesDestinationColumnNames(t *testing.T) { func TestDataBranchOutputAppenderAppendRowAndFlushAll(t *testing.T) { batchInfo := &applyBatchInfo{ - dbName: "db1", - baseTable: "t1", - deleteTable: "__mo_diff_del_x", - insertTable: "__mo_diff_ins_x", - deleteKeyNames: []string{"id"}, - writableNames: []string{"id", "name"}, + dbName: "db1", + baseTable: "t1", + deleteTable: "__mo_diff_del_x", + insertTable: "__mo_diff_ins_x", + deleteKeyNames: []string{"id"}, + deleteStageNames: []string{"branch_apply_key_0"}, + deleteKeyTypes: []types.Type{types.T_int64.ToType()}, + writableNames: []string{"id", "name"}, } t.Run("append delete in full-row mode", func(t *testing.T) { diff --git a/pkg/frontend/data_branch_pick.go b/pkg/frontend/data_branch_pick.go index f6250ce57bc4f..93009ad561c59 100644 --- a/pkg/frontend/data_branch_pick.go +++ b/pkg/frontend/data_branch_pick.go @@ -197,6 +197,11 @@ func pickMergeDiffs( ctx, ses, bh, tblStuff, dataBranchApplyModeOnlinePKOnly, &deleteCnt, deleteFromVals, &insertCnt, insertIntoVals, nil, ) + var acceptedExactFloatKeys map[string]struct{} + if appender.batchInfo.deleteNeedsExactFloatKeyMatch() && + stmt.ConflictOpt != nil && stmt.ConflictOpt.Opt == tree.CONFLICT_ACCEPT { + acceptedExactFloatKeys = make(map[string]struct{}) + } if err = initApplyTables(ctx, ses, bh, appender.batchInfo, appender.writeFile); err != nil { return err } @@ -218,7 +223,7 @@ func pickMergeDiffs( if err = appendPickedBatchRows( ctx, ses, tblStuff, wrapped, tmpValsBuffer, appender, - stmt.ConflictOpt, skipSet, + stmt.ConflictOpt, skipSet, acceptedExactFloatKeys, ); err != nil { firstErr = err cancel() @@ -228,6 +233,9 @@ func pickMergeDiffs( tblStuff.retPool.releaseRetBatch(wrapped.batch, false) } + if firstErr == nil && len(acceptedExactFloatKeys) != 0 { + firstErr = moerr.NewInternalErrorNoCtx("Data Branch PICK accepted conflict is missing its source row") + } if err = appender.flushAll(); err != nil { if firstErr == nil { @@ -267,6 +275,7 @@ func appendPickedBatchRows( appender sqlValuesAppender, userConflictOpt *tree.ConflictOpt, skipSet map[string]struct{}, + acceptedExactFloatKeys map[string]struct{}, ) (err error) { // PICK only cares about two kinds of batches from hashDiff: // 1. target INSERT (side=target, kind=INSERT) — source rows to add/replace @@ -279,6 +288,10 @@ func appendPickedBatchRows( if wrapped.side == diffSideBase && wrapped.kind != diffDelete { return nil } + exactFloatKeyUpdate, err := dataBranchExactFloatKeyUpdateBatch(wrapped, appender.batchInfo) + if err != nil { + return err + } row := make([]any, len(tblStuff.def.colNames)) @@ -310,7 +323,14 @@ func appendPickedBatchRows( skipSet[pkKey] = struct{}{} continue // do not apply the DELETE case tree.CONFLICT_ACCEPT: - // fall through — apply the DELETE (source value wins) + if acceptedExactFloatKeys != nil { + // The matching source row is applied by an exact-key upsert. + // Deferring the resolution avoids a staged delete that could + // run after the upsert and remove the accepted row. + acceptedExactFloatKeys[pkKey] = struct{}{} + continue + } + // Fall through for non-float keys: DELETE + INSERT is safe. } } } @@ -321,6 +341,15 @@ func appendPickedBatchRows( continue } } + if acceptedExactFloatKeys != nil && wrapped.side == diffSideTarget && wrapped.kind == diffInsert { + if _, accepted := acceptedExactFloatKeys[pkKey]; accepted { + exactFloatKeyUpdate = true + delete(acceptedExactFloatKeys, pkKey) + } + } + if exactFloatKeyUpdate && wrapped.kind == diffDelete { + continue + } if err = extractDataBranchApplyRow( ctx, ses, tblStuff, wrapped.batch, rowIdx, appender.extraColIdxesForRow(wrapped.kind), row, @@ -328,8 +357,9 @@ func appendPickedBatchRows( ); err != nil { return } - if err = appendDataBranchApplyRowAsSQLValues( + if err = appendOrExecuteDataBranchApplyRow( ctx, ses, tblStuff, wrapped.kind, row, tmpValsBuffer, appender, + exactFloatKeyUpdate, wrapped.restoreMissing, ); err != nil { return } diff --git a/pkg/frontend/data_branch_test.go b/pkg/frontend/data_branch_test.go index f935463bfeb31..694d35142b983 100644 --- a/pkg/frontend/data_branch_test.go +++ b/pkg/frontend/data_branch_test.go @@ -18,6 +18,7 @@ import ( "bytes" "context" "fmt" + "math" "net/url" "os" "path/filepath" @@ -28,6 +29,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/common/moerr" "github.com/matrixorigin/matrixone/pkg/common/mpool" "github.com/matrixorigin/matrixone/pkg/config" + "github.com/matrixorigin/matrixone/pkg/container/batch" "github.com/matrixorigin/matrixone/pkg/container/types" "github.com/matrixorigin/matrixone/pkg/container/vector" "github.com/matrixorigin/matrixone/pkg/defines" @@ -305,6 +307,63 @@ func TestFormatValIntoString_Nil(t *testing.T) { require.Equal(t, "NULL", buf.String()) } +func TestFormatValIntoString_FloatLiterals(t *testing.T) { + tests := []struct { + name string + val any + typ types.Type + want string + }{ + {"float32 finite", float32(1.25), types.T_float32.ToType(), "1.25"}, + {"float32 negative zero", math.Float32frombits(0x80000000), types.T_float32.ToType(), "bit_cast(unhex('00000080') as float)"}, + {"float32 NaN payload 0", math.Float32frombits(0x7fc00000), types.T_float32.ToType(), "bit_cast(unhex('0000c07f') as float)"}, + {"float32 NaN payload 1", math.Float32frombits(0x7fc00001), types.T_float32.ToType(), "bit_cast(unhex('0100c07f') as float)"}, + {"float32 positive infinity", float32(math.Inf(1)), types.T_float32.ToType(), "cast('+Inf' as float)"}, + {"float32 negative infinity", float32(math.Inf(-1)), types.T_float32.ToType(), "cast('-Inf' as float)"}, + {"float64 finite", 1.25, types.T_float64.ToType(), "1.25"}, + {"float64 negative zero", math.Float64frombits(0x8000000000000000), types.T_float64.ToType(), "bit_cast(unhex('0000000000000080') as double)"}, + {"float64 NaN payload 0", math.Float64frombits(0x7ff8000000000000), types.T_float64.ToType(), "bit_cast(unhex('000000000000f87f') as double)"}, + {"float64 NaN payload 1", math.Float64frombits(0x7ff8000000000001), types.T_float64.ToType(), "bit_cast(unhex('010000000000f87f') as double)"}, + {"float64 positive infinity", math.Inf(1), types.T_float64.ToType(), "cast('+Inf' as double)"}, + {"float64 negative infinity", math.Inf(-1), types.T_float64.ToType(), "cast('-Inf' as double)"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var buf bytes.Buffer + require.NoError(t, formatValIntoString(&Session{}, tt.val, tt.typ, &buf)) + require.Equal(t, tt.want, buf.String()) + }) + } +} + +func TestFormatValIntoStringWithFloatCast(t *testing.T) { + tests := []struct { + name string + val any + typ types.Type + want string + }{ + {"float32 finite", float32(1.25), types.T_float32.ToType(), "cast(1.25 as float)"}, + {"float32 NaN", math.Float32frombits(0x7fc00001), types.T_float32.ToType(), "bit_cast(unhex('0100c07f') as float)"}, + {"float32 negative zero", math.Float32frombits(0x80000000), types.T_float32.ToType(), "bit_cast(unhex('00000080') as float)"}, + {"float64 finite", 1.25, types.T_float64.ToType(), "cast(1.25 as double)"}, + {"float64 infinity", math.Inf(1), types.T_float64.ToType(), "cast('+Inf' as double)"}, + {"float64 negative zero", math.Float64frombits(0x8000000000000000), types.T_float64.ToType(), "bit_cast(unhex('0000000000000080') as double)"}, + {"non-float unchanged", int64(7), types.T_int64.ToType(), "7"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var buf bytes.Buffer + require.NoError(t, formatValIntoStringWithFloatCast( + &Session{}, tt.val, tt.typ, &buf, true, + )) + require.Equal(t, tt.want, buf.String()) + }) + } +} + func TestFormatValIntoString_DataBranchSpecialTypes(t *testing.T) { tests := []struct { name string @@ -960,6 +1019,97 @@ func TestCompareSingleValInVector_ConstVectors(t *testing.T) { require.Equal(t, types.CompareValue(int32(5), int32(7)), cmp) } +func TestDataBranchPrimaryKeyFloatIdentity(t *testing.T) { + ctx := context.Background() + ses := &Session{} + mp := mpool.MustNewZero() + defer mpool.DeleteMPool(mp) + + float32Vec := buildFixedVector(t, mp, types.T_float32.ToType(), + math.Float32frombits(0x00000000), + math.Float32frombits(0x80000000), + ) + defer float32Vec.Free(mp) + cmp, err := compareDataBranchPrimaryKeyInVectors(ctx, ses, 0, 1, float32Vec, float32Vec) + require.NoError(t, err) + require.Equal(t, -1, cmp) + cmp, err = compareDataBranchPrimaryKeyInVectors(ctx, ses, 1, 0, float32Vec, float32Vec) + require.NoError(t, err) + require.Equal(t, 1, cmp) + + float64Vec := buildFixedVector(t, mp, types.T_float64.ToType(), + math.Float64frombits(0x7ff8000000000000), + math.Float64frombits(0x7ff8000000000001), + ) + defer float64Vec.Free(mp) + cmp, err = compareDataBranchPrimaryKeyInVectors(ctx, ses, 0, 1, float64Vec, float64Vec) + require.NoError(t, err) + require.Equal(t, -1, cmp) + cmp, err = compareDataBranchPrimaryKeyInVectors(ctx, ses, 1, 1, float64Vec, float64Vec) + require.NoError(t, err) + require.Zero(t, cmp) + + constFloat, err := vector.NewConstFixed[float64]( + types.T_float64.ToType(), math.Float64frombits(0x8000000000000000), 2, mp, + ) + require.NoError(t, err) + defer constFloat.Free(mp) + identity, isNull, err := dataBranchFloatPKIdentityAt(constFloat, 1) + require.NoError(t, err) + require.False(t, isNull) + require.Equal(t, uint64(0x8000000000000000), identity) + + nullFloat := vector.NewConstNull(types.T_float64.ToType(), 1, mp) + defer nullFloat.Free(mp) + cmp, err = compareDataBranchPrimaryKeyInVectors(ctx, ses, 0, 0, nullFloat, float64Vec) + require.NoError(t, err) + require.Equal(t, -1, cmp) + cmp, err = compareDataBranchPrimaryKeyInVectors(ctx, ses, 0, 0, float64Vec, nullFloat) + require.NoError(t, err) + require.Equal(t, 1, cmp) + cmp, err = compareDataBranchPrimaryKeyInVectors(ctx, ses, 0, 0, nullFloat, nullFloat) + require.NoError(t, err) + require.Zero(t, cmp) + + intVec := buildFixedVector(t, mp, types.T_int64.ToType(), int64(1), int64(2)) + defer intVec.Free(mp) + cmp, err = compareDataBranchPrimaryKeyInVectors(ctx, ses, 0, 1, intVec, intVec) + require.NoError(t, err) + require.Equal(t, -1, cmp) + _, _, err = dataBranchFloatPKIdentityAt(intVec, 0) + require.ErrorContains(t, err, "requires FLOAT/DOUBLE") +} + +func TestSortDataBranchBatchByExactFloatPrimaryKey(t *testing.T) { + mp := mpool.MustNewZero() + defer mpool.DeleteMPool(mp) + + bat := batch.NewWithSize(2) + bat.Vecs[0] = buildFixedVector(t, mp, types.T_float64.ToType(), + math.Float64frombits(0x7ff8000000000001), + math.Float64frombits(0x0000000000000000), + math.Float64frombits(0x7ff8000000000000), + math.Float64frombits(0x8000000000000000), + ) + bat.Vecs[1] = buildFixedVector(t, mp, types.T_int64.ToType(), int64(1), int64(2), int64(3), int64(4)) + bat.SetRowCount(4) + defer bat.Clean(mp) + + require.NoError(t, sortDataBranchBatchByPrimaryKey(bat, 0, mp)) + require.Equal(t, []uint64{ + 0x0000000000000000, + 0x7ff8000000000000, + 0x7ff8000000000001, + 0x8000000000000000, + }, []uint64{ + math.Float64bits(vector.GetFixedAtNoTypeCheck[float64](bat.Vecs[0], 0)), + math.Float64bits(vector.GetFixedAtNoTypeCheck[float64](bat.Vecs[0], 1)), + math.Float64bits(vector.GetFixedAtNoTypeCheck[float64](bat.Vecs[0], 2)), + math.Float64bits(vector.GetFixedAtNoTypeCheck[float64](bat.Vecs[0], 3)), + }) + require.Equal(t, []int64{2, 3, 1, 4}, vector.MustFixedColWithTypeCheck[int64](bat.Vecs[1])) +} + func TestCompareTupleValueWithVectorDecimal256(t *testing.T) { mp := mpool.MustNewZero() defer mpool.DeleteMPool(mp) diff --git a/pkg/frontend/data_branch_types.go b/pkg/frontend/data_branch_types.go index 8279e246a4388..338d740b2c96e 100644 --- a/pkg/frontend/data_branch_types.go +++ b/pkg/frontend/data_branch_types.go @@ -270,11 +270,12 @@ func (t *tableStuff) resolvedSnapshots(ses *Session) (tarSP, baseSP types.TS) { } type batchWithKind struct { - name string - kind string - side int - fromUpdate bool - batch *batch.Batch + name string + kind string + side int + fromUpdate bool + restoreMissing bool + batch *batch.Batch } type emitFunc func(batchWithKind) (stop bool, err error) diff --git a/pkg/sql/colexec/dedup_key.go b/pkg/sql/colexec/dedup_key.go new file mode 100644 index 0000000000000..89e07397c7ac3 --- /dev/null +++ b/pkg/sql/colexec/dedup_key.go @@ -0,0 +1,48 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package colexec + +import ( + "strings" + + "github.com/matrixorigin/matrixone/pkg/container/types" + "github.com/matrixorigin/matrixone/pkg/container/vector" + "github.com/matrixorigin/matrixone/pkg/pb/plan" +) + +// FormatDedupKey formats a duplicate key from the expression vector used by a +// DEDUP join. FLOAT/DOUBLE primary-key identity expressions are serial(...) +// encodings, so decode those bytes with the original column types instead of +// leaking the binary identity key into the user-facing duplicate-entry error. +func FormatDedupKey(vec *vector.Vector, row int, colTypes []plan.Type) (string, error) { + if len(colTypes) == 1 { + originalType := types.T(colTypes[0].Id) + if (originalType == types.T_float32 || originalType == types.T_float64) && + (vec.GetType().Oid == types.T_varchar || vec.GetType().Oid == types.T_varbinary) { + items, err := types.StringifyTuple(vec.GetBytesAt(row), colTypes) + if err != nil { + return "", err + } + return items[0], nil + } + return vec.RowToString(row), nil + } + + items, err := types.StringifyTuple(vec.GetBytesAt(row), colTypes) + if err != nil { + return "", err + } + return "(" + strings.Join(items, ",") + ")", nil +} diff --git a/pkg/sql/colexec/dedup_key_test.go b/pkg/sql/colexec/dedup_key_test.go new file mode 100644 index 0000000000000..eeac081b52e06 --- /dev/null +++ b/pkg/sql/colexec/dedup_key_test.go @@ -0,0 +1,68 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package colexec + +import ( + "math" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/matrixorigin/matrixone/pkg/common/mpool" + "github.com/matrixorigin/matrixone/pkg/container/types" + "github.com/matrixorigin/matrixone/pkg/container/vector" + "github.com/matrixorigin/matrixone/pkg/pb/plan" +) + +func TestFormatDedupKeyDecodesFloatIdentity(t *testing.T) { + pool := mpool.MustNewZero() + packer := types.NewPacker() + defer packer.Close() + + packer.EncodeFloat32(1) + float32Key := append([]byte(nil), packer.Bytes()...) + packer.Reset() + packer.EncodeFloat64(math.Copysign(0, -1)) + float64Key := append([]byte(nil), packer.Bytes()...) + + vec := vector.NewVec(types.T_varchar.ToType()) + defer vec.Free(pool) + require.NoError(t, vector.AppendBytes(vec, float32Key, false, pool)) + require.NoError(t, vector.AppendBytes(vec, float64Key, false, pool)) + + got, err := FormatDedupKey(vec, 0, []plan.Type{{Id: int32(types.T_float32)}}) + require.NoError(t, err) + require.Equal(t, "1", got) + + got, err = FormatDedupKey(vec, 1, []plan.Type{{Id: int32(types.T_float64)}}) + require.NoError(t, err) + require.Equal(t, "-0", got) +} + +func TestFormatDedupKeyRejectsMalformedEncodedIdentity(t *testing.T) { + pool := mpool.MustNewZero() + vec := vector.NewVec(types.T_varchar.ToType()) + defer vec.Free(pool) + require.NoError(t, vector.AppendBytes(vec, []byte{0xff}, false, pool)) + + _, err := FormatDedupKey(vec, 0, []plan.Type{{Id: int32(types.T_float64)}}) + require.Error(t, err) + + _, err = FormatDedupKey(vec, 0, []plan.Type{ + {Id: int32(types.T_float64)}, + {Id: int32(types.T_int64)}, + }) + require.Error(t, err) +} diff --git a/pkg/sql/colexec/dedupjoin/join.go b/pkg/sql/colexec/dedupjoin/join.go index d6dbebb91babf..6daa02e54d106 100644 --- a/pkg/sql/colexec/dedupjoin/join.go +++ b/pkg/sql/colexec/dedupjoin/join.go @@ -16,7 +16,6 @@ package dedupjoin import ( "bytes" "context" - "strings" "github.com/matrixorigin/matrixone/pkg/catalog" "github.com/matrixorigin/matrixone/pkg/common/hashmap" @@ -941,14 +940,16 @@ func (ctr *container) probe(bat *batch.Batch, ap *DedupJoin, proc *process.Proce } } if len(rowStr) == 0 { - rowStr = ctr.vecs[0].RowToString(i + k) + rowStr, err = colexec.FormatDedupKey(ctr.vecs[0], i+k, ap.DedupColTypes) + if err != nil { + return err + } } } else { - rowItems, err := types.StringifyTuple(ctr.vecs[0].GetBytesAt(i+k), ap.DedupColTypes) + rowStr, err = colexec.FormatDedupKey(ctr.vecs[0], i+k, ap.DedupColTypes) if err != nil { return err } - rowStr = "(" + strings.Join(rowItems, ",") + ")" } return moerr.NewDuplicateEntry(proc.Ctx, rowStr, ap.DedupColName) case plan.Node_IGNORE: diff --git a/pkg/sql/colexec/hashbuild/hashmap.go b/pkg/sql/colexec/hashbuild/hashmap.go index 58f524b56b978..a5ca147f60759 100644 --- a/pkg/sql/colexec/hashbuild/hashmap.go +++ b/pkg/sql/colexec/hashbuild/hashmap.go @@ -17,7 +17,6 @@ package hashbuild import ( "math" "runtime" - "strings" "github.com/matrixorigin/matrixone/pkg/catalog" "github.com/matrixorigin/matrixone/pkg/common/bitmap" @@ -744,14 +743,16 @@ buildUnits: } if len(rowStr) == 0 { - rowStr = hb.curVecs[0].RowToString(vecIdx2 + k) + rowStr, err = colexec.FormatDedupKey(hb.curVecs[0], vecIdx2+k, hb.DedupColTypes) + if err != nil { + return err + } } } else { - rowItems, err := types.StringifyTuple(hb.curVecs[0].GetBytesAt(vecIdx2+k), hb.DedupColTypes) + rowStr, err = colexec.FormatDedupKey(hb.curVecs[0], vecIdx2+k, hb.DedupColTypes) if err != nil { return err } - rowStr = "(" + strings.Join(rowItems, ",") + ")" } return moerr.NewDuplicateEntry(proc.Ctx, rowStr, hb.DedupColName) case plan.Node_IGNORE: diff --git a/pkg/sql/colexec/rightdedupjoin/join.go b/pkg/sql/colexec/rightdedupjoin/join.go index 27b9f3225cc61..3e1baf50b705a 100644 --- a/pkg/sql/colexec/rightdedupjoin/join.go +++ b/pkg/sql/colexec/rightdedupjoin/join.go @@ -16,7 +16,6 @@ package rightdedupjoin import ( "bytes" - "strings" "github.com/matrixorigin/matrixone/pkg/catalog" "github.com/matrixorigin/matrixone/pkg/common/hashmap" @@ -374,14 +373,16 @@ func (ctr *container) probe(bat *batch.Batch, ap *RightDedupJoin, proc *process. } if len(rowStr) == 0 { - rowStr = ctr.vecs[0].RowToString(i + k) + rowStr, err = colexec.FormatDedupKey(ctr.vecs[0], i+k, ap.DedupColTypes) + if err != nil { + return err + } } } else { - rowItems, err := types.StringifyTuple(ctr.vecs[0].GetBytesAt(i+k), ap.DedupColTypes) + rowStr, err = colexec.FormatDedupKey(ctr.vecs[0], i+k, ap.DedupColTypes) if err != nil { return err } - rowStr = "(" + strings.Join(rowItems, ",") + ")" } return moerr.NewDuplicateEntry(proc.Ctx, rowStr, ap.DedupColName) } diff --git a/pkg/sql/compile/fuzzyCheck.go b/pkg/sql/compile/fuzzyCheck.go index fa515b6f6d82f..fd009b61cb29d 100644 --- a/pkg/sql/compile/fuzzyCheck.go +++ b/pkg/sql/compile/fuzzyCheck.go @@ -83,6 +83,11 @@ func newFuzzyCheck(node *plan.Node) (*fuzzyCheck, error) { } } + if !f.isCompound && f.col != nil { + colType := types.T(f.col.Typ.Id) + f.exactFloatKey = colType == types.T_float32 || colType == types.T_float64 + } + return f, nil } @@ -101,6 +106,7 @@ func (f *fuzzyCheck) clear() { f.attr = "" f.condition = "" f.isCompound = false + f.exactFloatKey = false f.onlyInsertHidden = false f.col = nil f.compoundCols = nil @@ -225,7 +231,23 @@ func (f *fuzzyCheck) fill(ctx context.Context, bat *batch.Batch) error { func (f *fuzzyCheck) firstlyCheck(ctx context.Context, toCheck *vector.Vector) error { kcnt := make(map[string]int) - if !f.isCompound { + if f.exactFloatKey { + for i := 0; i < toCheck.Length(); i++ { + if toCheck.GetNulls().Contains(uint64(i)) { + continue + } + key := string(toCheck.GetBytesAt(i)) + kcnt[key]++ + if kcnt[key] > 1 { + display, err := f.exactFloatKeyDisplay([]byte(key)) + if err != nil { + return err + } + return moerr.NewDuplicateEntry(ctx, display, f.attr) + } + } + return nil + } else if !f.isCompound { pkey, err := f.format(toCheck) if err != nil { return err @@ -284,7 +306,13 @@ func (f *fuzzyCheck) genCollsionKeys(toCheck *vector.Vector) ([][]string, error) } if !f.onlyInsertHidden { - if !f.isCompound { + if f.exactFloatKey { + for i := 0; i < toCheck.Length(); i++ { + if !toCheck.GetNulls().Contains(uint64(i)) { + keys[0] = append(keys[0], "unhex('"+hex.EncodeToString(toCheck.GetBytesAt(i))+"')") + } + } + } else if !f.isCompound { pkey, err := f.format(toCheck) if err != nil { return nil, err @@ -363,22 +391,7 @@ func fuzzyCheckSQLValueNeedsQuote(typ types.T) bool { // backgroundSQLCheck launches a background SQL to check if there are any duplicates func (f *fuzzyCheck) backgroundSQLCheck(c *Compile) error { - var duplicateCheckSql string - - if !f.onlyInsertHidden { - if !f.isCompound { - duplicateCheckSql = fmt.Sprintf(fuzzyNonCompoundCheck, f.attr, f.db, f.tbl, f.attr, f.condition, f.attr) - } else { - cAttrs := make([]string, len(f.compoundCols)) - for k, c := range f.compoundCols { - cAttrs[k] = c.Name - } - attrs := strings.Join(cAttrs, ", ") - duplicateCheckSql = fmt.Sprintf(fuzzyCompoundCheck, attrs, f.db, f.tbl, f.condition, attrs) - } - } else { - duplicateCheckSql = fmt.Sprintf(fuzzyNonCompoundCheck, f.attr, f.db, f.tbl, f.attr, f.condition, f.attr) - } + duplicateCheckSql := f.duplicateCheckSQL() res, err := c.runSqlWithResultAndOptions(duplicateCheckSql, NoAccountId, executor.StatementOption{}.WithDisableLog()) if err != nil { @@ -393,15 +406,24 @@ func (f *fuzzyCheck) backgroundSQLCheck(c *Compile) error { if vs != nil && vs[0].Length() > 0 { // do dup toCheck := vs[0] if !f.isCompound { - f.adjustDecimalScale(toCheck) - if dupKey, e := f.format(toCheck); e != nil { - err = e - } else { - ds, e := strconv.Unquote(dupKey[0]) + if f.exactFloatKey { + display, e := f.exactFloatKeyDisplay(toCheck.GetBytesAt(0)) if e != nil { - err = moerr.NewDuplicateEntry(c.proc.Ctx, dupKey[0], f.attr) + err = e + } else { + err = moerr.NewDuplicateEntry(c.proc.Ctx, display, f.attr) + } + } else { + f.adjustDecimalScale(toCheck) + if dupKey, e := f.format(toCheck); e != nil { + err = e } else { - err = moerr.NewDuplicateEntry(c.proc.Ctx, ds, f.attr) + ds, e := strconv.Unquote(dupKey[0]) + if e != nil { + err = moerr.NewDuplicateEntry(c.proc.Ctx, dupKey[0], f.attr) + } else { + err = moerr.NewDuplicateEntry(c.proc.Ctx, ds, f.attr) + } } } } else { @@ -425,6 +447,37 @@ func (f *fuzzyCheck) backgroundSQLCheck(c *Compile) error { return err } +func (f *fuzzyCheck) duplicateCheckSQL() string { + var duplicateCheckSql string + + if !f.onlyInsertHidden { + if f.exactFloatKey { + identityExpr := fmt.Sprintf("serial(%s)", f.attr) + duplicateCheckSql = fmt.Sprintf(fuzzyNonCompoundCheck, identityExpr, f.db, f.tbl, identityExpr, f.condition, identityExpr) + } else if !f.isCompound { + duplicateCheckSql = fmt.Sprintf(fuzzyNonCompoundCheck, f.attr, f.db, f.tbl, f.attr, f.condition, f.attr) + } else { + cAttrs := make([]string, len(f.compoundCols)) + for k, c := range f.compoundCols { + cAttrs[k] = c.Name + } + attrs := strings.Join(cAttrs, ", ") + duplicateCheckSql = fmt.Sprintf(fuzzyCompoundCheck, attrs, f.db, f.tbl, f.condition, attrs) + } + } else { + duplicateCheckSql = fmt.Sprintf(fuzzyNonCompoundCheck, f.attr, f.db, f.tbl, f.attr, f.condition, f.attr) + } + return duplicateCheckSql +} + +func (f *fuzzyCheck) exactFloatKeyDisplay(key []byte) (string, error) { + values, err := types.StringifyTuple(key, []plan.Type{f.col.Typ}) + if err != nil { + return "", err + } + return values[0], nil +} + // -----------------------------utils----------------------------------- // make sure that the attr sort by define way diff --git a/pkg/sql/compile/fuzzyCheck_test.go b/pkg/sql/compile/fuzzyCheck_test.go index 7f5fa777d48d8..379eda5b7b5aa 100644 --- a/pkg/sql/compile/fuzzyCheck_test.go +++ b/pkg/sql/compile/fuzzyCheck_test.go @@ -16,19 +16,158 @@ package compile import ( "context" + "encoding/hex" + "math" + "strings" "sync" "testing" "time" "github.com/matrixorigin/matrixone/pkg/common/mpool" + moruntime "github.com/matrixorigin/matrixone/pkg/common/runtime" "github.com/matrixorigin/matrixone/pkg/common/sqlquote" "github.com/matrixorigin/matrixone/pkg/container/batch" "github.com/matrixorigin/matrixone/pkg/container/types" "github.com/matrixorigin/matrixone/pkg/container/vector" "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/testutil" + "github.com/matrixorigin/matrixone/pkg/util/executor" "github.com/stretchr/testify/require" ) +func TestExactFloatFuzzyCheckPreservesStorageIdentity(t *testing.T) { + mp, err := mpool.NewMPool("test_exact_float_fuzzy_check", 0, mpool.NoFixed) + require.NoError(t, err) + defer mpool.DeleteMPool(mp) + + packer := types.NewPacker() + defer packer.Close() + encode := func(value float64) []byte { + packer.Reset() + packer.EncodeFloat64(value) + return append([]byte(nil), packer.GetBuf()...) + } + + positiveZero := encode(0) + negativeZero := encode(math.Copysign(0, -1)) + nanPayload0 := encode(math.Float64frombits(0x7ff8000000000000)) + nanPayload1 := encode(math.Float64frombits(0x7ff8000000000001)) + + vec := vector.NewVec(types.T_varchar.ToType()) + defer vec.Free(mp) + for _, key := range [][]byte{positiveZero, negativeZero, nanPayload0, nanPayload1} { + require.NoError(t, vector.AppendBytes(vec, key, false, mp)) + } + + f := &fuzzyCheck{ + db: "db", + tbl: "t", + attr: "k", + exactFloatKey: true, + col: &plan.ColDef{ + Name: "k", + Typ: plan.Type{Id: int32(types.T_float64)}, + }, + } + require.NoError(t, f.firstlyCheck(context.Background(), vec), + "signed zero and distinct NaN payloads are distinct primary keys") + + keys, err := f.genCollsionKeys(vec) + require.NoError(t, err) + require.Equal(t, []string{ + "unhex('" + hex.EncodeToString(positiveZero) + "')", + "unhex('" + hex.EncodeToString(negativeZero) + "')", + "unhex('" + hex.EncodeToString(nanPayload0) + "')", + "unhex('" + hex.EncodeToString(nanPayload1) + "')", + }, keys[0]) + f.condition = strings.Join(keys[0], ", ") + require.Equal(t, + "select serial(k) from `db`.`t` where serial(k) in ("+f.condition+") group by serial(k) having count(*) > 1 limit 1;", + f.duplicateCheckSQL()) + + require.NoError(t, vector.AppendBytes(vec, negativeZero, false, mp)) + err = f.firstlyCheck(context.Background(), vec) + require.ErrorContains(t, err, "Duplicate entry '-0'") + + _, err = f.exactFloatKeyDisplay([]byte{0xff}) + require.Error(t, err) +} + +func TestFuzzyCheckDuplicateSQLModes(t *testing.T) { + compound := &fuzzyCheck{ + db: "db", + tbl: "t", + condition: "(a = 1 and b = 2)", + isCompound: true, + compoundCols: []*plan.ColDef{{Name: "a"}, {Name: "b"}}, + } + require.Equal(t, + "select serial(a, b) from `db`.`t` where (a = 1 and b = 2) group by serial(a, b) having count(*) > 1 limit 1;", + compound.duplicateCheckSQL()) + + hidden := &fuzzyCheck{ + db: "db", + tbl: "hidden", + attr: "k", + condition: "1, 2", + onlyInsertHidden: true, + } + require.Equal(t, + "select k from `db`.`hidden` where k in (1, 2) group by k having count(*) > 1 limit 1;", + hidden.duplicateCheckSQL()) +} + +func TestExactFloatBackgroundSQLCheckFormatsDuplicateIdentity(t *testing.T) { + proc := testutil.NewProcess(t) + compile := &Compile{ + proc: proc, + pn: &plan.Plan{Plan: &plan.Plan_Query{Query: &plan.Query{}}}, + } + f := &fuzzyCheck{ + db: "db", + tbl: "t", + attr: "k", + exactFloatKey: true, + col: &plan.ColDef{ + Name: "k", + Typ: plan.Type{Id: int32(types.T_float64)}, + }, + } + + check := func(t *testing.T, key []byte) error { + memResult := executor.NewMemResult([]types.Type{types.T_varchar.ToType()}, proc.Mp()) + memResult.NewBatchWithRowCount(1) + require.NoError(t, executor.AppendBytesRows(memResult, 0, [][]byte{key})) + rt := moruntime.ServiceRuntime(proc.GetService()) + oldExecutor, hadOldExecutor := rt.GetGlobalVariables(moruntime.InternalSQLExecutor) + newExecutor := executor.NewMemExecutor(func(sql string) (executor.Result, error) { + require.Equal(t, f.duplicateCheckSQL(), sql) + return memResult.GetResult(), nil + }) + rt.SetGlobalVariables(moruntime.InternalSQLExecutor, newExecutor) + t.Cleanup(func() { + if hadOldExecutor { + rt.SetGlobalVariables(moruntime.InternalSQLExecutor, oldExecutor) + } else { + rt.CompareAndDeleteGlobalVariables(moruntime.InternalSQLExecutor, newExecutor) + } + }) + return f.backgroundSQLCheck(compile) + } + + t.Run("decoded duplicate", func(t *testing.T) { + packer := types.NewPacker() + defer packer.Close() + packer.EncodeFloat64(math.Copysign(0, -1)) + err := check(t, append([]byte(nil), packer.GetBuf()...)) + require.ErrorContains(t, err, "Duplicate entry '-0'") + }) + + t.Run("malformed identity", func(t *testing.T) { + require.Error(t, check(t, []byte{0xff})) + }) +} + func TestVectorToStringNullHandling(t *testing.T) { mp, err := mpool.NewMPool("test_vectorToString", 0, mpool.NoFixed) require.NoError(t, err) diff --git a/pkg/sql/compile/operator.go b/pkg/sql/compile/operator.go index 1a0791c4aa31a..79c3ea7baa40f 100644 --- a/pkg/sql/compile/operator.go +++ b/pkg/sql/compile/operator.go @@ -716,6 +716,14 @@ func constructFuzzyFilter(node, tableScan, sinkScan *plan.Node) *fuzzyfilter.Fuz } } } + // The fuzzy-filter children may project a key identity expression whose + // type differs from the stored column. FLOAT/DOUBLE primary keys use + // serial(...) bytes, and the operator must allocate/hash that actual type. + if len(sinkScan.ProjectList) > 0 { + pkTyp = sinkScan.ProjectList[0].Typ + } else if len(tableScan.ProjectList) > 0 { + pkTyp = tableScan.ProjectList[0].Typ + } op := fuzzyfilter.NewArgument() op.PkName = pkName diff --git a/pkg/sql/compile/operator_test.go b/pkg/sql/compile/operator_test.go index ba15db4ea7119..dcf717fea0b35 100644 --- a/pkg/sql/compile/operator_test.go +++ b/pkg/sql/compile/operator_test.go @@ -146,6 +146,32 @@ func TestConstructFuzzyFilterUsesFinalizedBuildSide(t *testing.T) { require.Empty(t, node.RuntimeFilterBuildList) require.Empty(t, tableScan.RuntimeFilterProbeList) }) + + t.Run("uses projected exact float identity type", func(t *testing.T) { + node, tableScan, sinkScan, _ := newNodes( + plan.Node_FUZZY_BUILD_SIDE_SINK, 10, 10) + node.TableDef.Cols[0].Typ = plan.Type{Id: int32(types.T_float64)} + identityType := plan.Type{Id: int32(types.T_varchar)} + tableScan.ProjectList = []*plan.Expr{{Typ: identityType}} + sinkScan.ProjectList = []*plan.Expr{{Typ: identityType}} + + op := constructFuzzyFilter(node, tableScan, sinkScan) + defer op.Release() + + require.Equal(t, identityType, op.PkTyp) + }) + + t.Run("uses table projection when sink projection is absent", func(t *testing.T) { + node, tableScan, sinkScan, _ := newNodes( + plan.Node_FUZZY_BUILD_SIDE_TABLE, 10, 10) + identityType := plan.Type{Id: int32(types.T_varchar)} + tableScan.ProjectList = []*plan.Expr{{Typ: identityType}} + + op := constructFuzzyFilter(node, tableScan, sinkScan) + defer op.Release() + + require.Equal(t, identityType, op.PkTyp) + }) } func TestConstructAggregateConfigIncludesGroupConcatMaxLen(t *testing.T) { diff --git a/pkg/sql/compile/types.go b/pkg/sql/compile/types.go index c4365a7b82908..bbed555d09af1 100644 --- a/pkg/sql/compile/types.go +++ b/pkg/sql/compile/types.go @@ -383,6 +383,10 @@ type fuzzyCheck struct { // handle with primary key(a, b, ...) or unique key (a, b, ...) isCompound bool + // exactFloatKey means the pipeline carries serial(FLOAT/DOUBLE) rather than + // the scalar key. This preserves signed zero and NaN payload identity. + exactFloatKey bool + // handle with cases like create a unique index for existed table, or alter add unique key // and the type of unique key is compound onlyInsertHidden bool diff --git a/pkg/sql/plan/bind_insert.go b/pkg/sql/plan/bind_insert.go index e4c77ff3319d9..5ae9fe97a456d 100644 --- a/pkg/sql/plan/bind_insert.go +++ b/pkg/sql/plan/bind_insert.go @@ -2127,16 +2127,26 @@ func (builder *QueryBuilder) appendDedupAndMultiUpdateNodesForBindInsert( if useTargetPk { rightPkPos = targetPkPos } - joinCond, _ = BindFuncExprImplByPlanExpr(builder.GetContext(), "=", []*plan.Expr{ - { - Typ: pkTyp, - Expr: &plan.Expr_Col{Col: &plan.ColRef{RelPos: scanTag, ColPos: pkPos}}, - }, - { - Typ: pkTyp, - Expr: &plan.Expr_Col{Col: &plan.ColRef{RelPos: selectTag, ColPos: rightPkPos}}, - }, - }) + leftPK := &plan.Expr{ + Typ: pkTyp, + Expr: &plan.Expr_Col{Col: &plan.ColRef{RelPos: scanTag, ColPos: pkPos}}, + } + rightPK := &plan.Expr{ + Typ: pkTyp, + Expr: &plan.Expr_Col{Col: &plan.ColRef{RelPos: selectTag, ColPos: rightPkPos}}, + } + leftPK, err = bindPrimaryKeyIdentityExpr(builder, leftPK, pkTyp) + if err != nil { + return 0, err + } + rightPK, err = bindPrimaryKeyIdentityExpr(builder, rightPK, pkTyp) + if err != nil { + return 0, err + } + joinCond, err = BindFuncExprImplByPlanExpr(builder.GetContext(), "=", []*plan.Expr{leftPK, rightPK}) + if err != nil { + return 0, err + } } var dedupColName string diff --git a/pkg/sql/plan/build_constraint_util.go b/pkg/sql/plan/build_constraint_util.go index 058c6d7b70ddd..c2b2532ab9b7c 100644 --- a/pkg/sql/plan/build_constraint_util.go +++ b/pkg/sql/plan/build_constraint_util.go @@ -1986,6 +1986,10 @@ func appendPrimaryConstraintPlan( }, }, } + pkColExpr, err = bindPrimaryKeyIdentityExpr(builder, pkColExpr, pkTyp) + if err != nil { + return err + } lastNodeId, err = appendAggCountGroupByColExpr(builder, bindCtx, lastNodeId, pkColExpr) if err != nil { return err @@ -2005,13 +2009,25 @@ func appendPrimaryConstraintPlan( if err != nil { return err } - varcharType := types.T_varchar.ToType() - varcharExpr, err := makePlan2CastExpr(builder.GetContext(), &Expr{ - Typ: tableDef.Cols[pkPos].Typ, + // The group key is the exact identity expression. FLOAT/DOUBLE keys + // therefore arrive here as serial(...) bytes; recover the original + // value for the user-facing duplicate-entry message without changing + // the bit-preserving grouping semantics. + displayExpr := &Expr{ + Typ: pkColExpr.Typ, Expr: &plan.Expr_Col{ Col: &plan.ColRef{ColPos: 1, Name: tableDef.Cols[pkPos].Name}, }, - }, makePlan2Type(&varcharType)) + } + pkType := types.T(pkTyp.Id) + if pkType == types.T_float32 || pkType == types.T_float64 { + displayExpr, err = MakeSerialExtractExpr(builder.GetContext(), displayExpr, pkTyp, 0) + if err != nil { + return err + } + } + varcharType := types.T_varchar.ToType() + varcharExpr, err := makePlan2CastExpr(builder.GetContext(), displayExpr, makePlan2Type(&varcharType)) if err != nil { return err } @@ -2048,23 +2064,30 @@ func appendPrimaryConstraintPlan( }, }, } - // sink_scan - sinkScanNode := &Node{ - NodeType: plan.Node_SINK_SCAN, - Stats: &plan.Stats{}, - SourceStep: []int32{sourceStep}, - ProjectList: []*Expr{ - &plan.Expr{ - Typ: pkTyp, - Expr: &plan.Expr_Col{ - Col: &plan.ColRef{ - ColPos: int32(pkPos), - Name: tableDef.Pkey.PkeyColName, - }, - }, + probeExpr, err = bindPrimaryKeyIdentityExpr(builder, probeExpr, pkTyp) + if err != nil { + return err + } + sourcePKExpr := &plan.Expr{ + Typ: pkTyp, + Expr: &plan.Expr_Col{ + Col: &plan.ColRef{ + ColPos: int32(pkPos), + Name: tableDef.Pkey.PkeyColName, }, }, } + sourcePKExpr, err = bindPrimaryKeyIdentityExpr(builder, sourcePKExpr, pkTyp) + if err != nil { + return err + } + // sink_scan + sinkScanNode := &Node{ + NodeType: plan.Node_SINK_SCAN, + Stats: &plan.Stats{}, + SourceStep: []int32{sourceStep}, + ProjectList: []*Expr{sourcePKExpr}, + } lastNodeId = builder.appendNode(sinkScanNode, bindCtx) pkNameMap := make(map[string]int) @@ -2086,20 +2109,25 @@ func appendPrimaryConstraintPlan( } } - scanNode := &plan.Node{ - NodeType: plan.Node_TABLE_SCAN, - Stats: &plan.Stats{}, - ObjRef: objRef, - TableDef: scanTableDef, - ProjectList: []*Expr{{ - Typ: pkTyp, - Expr: &plan.Expr_Col{ - Col: &ColRef{ - ColPos: int32(len(scanTableDef.Cols) - 1), - Name: tableDef.Pkey.PkeyColName, - }, + scanPKExpr := &plan.Expr{ + Typ: pkTyp, + Expr: &plan.Expr_Col{ + Col: &ColRef{ + ColPos: int32(len(scanTableDef.Cols) - 1), + Name: tableDef.Pkey.PkeyColName, }, - }}, + }, + } + scanPKExpr, err = bindPrimaryKeyIdentityExpr(builder, scanPKExpr, pkTyp) + if err != nil { + return err + } + scanNode := &plan.Node{ + NodeType: plan.Node_TABLE_SCAN, + Stats: &plan.Stats{}, + ObjRef: objRef, + TableDef: scanTableDef, + ProjectList: []*Expr{scanPKExpr}, } if builder.isRestore { @@ -2140,7 +2168,7 @@ func appendPrimaryConstraintPlan( if len(pkFilterExprs) == 0 { buildExpr := &plan.Expr{ - Typ: pkTyp, + Typ: scanPKExpr.Typ, Expr: &plan.Expr_Col{ Col: &plan.ColRef{ RelPos: 0, @@ -2445,3 +2473,14 @@ func appendPrimaryConstraintPlan( return nil } + +func bindPrimaryKeyIdentityExpr(builder *QueryBuilder, expr *Expr, typ plan.Type) (*Expr, error) { + pkType := types.T(typ.Id) + if pkType != types.T_float32 && pkType != types.T_float64 { + return expr, nil + } + // FLOAT/DOUBLE primary-key identity is bit-preserving. Feed serial() + // encodings to duplicate-check paths so NaN payloads and signed zero are + // neither collapsed nor matched inconsistently by scalar-key hash joins. + return BindFuncExprImplByPlanExpr(builder.GetContext(), "serial", []*Expr{expr}) +} diff --git a/test/distributed/cases/git4data/branch/edge/branch_float_special_values.result b/test/distributed/cases/git4data/branch/edge/branch_float_special_values.result new file mode 100644 index 0000000000000..6368790cd176d --- /dev/null +++ b/test/distributed/cases/git4data/branch/edge/branch_float_special_values.result @@ -0,0 +1,543 @@ +drop database if exists br_float_special_values; +create database br_float_special_values; +use br_float_special_values; +create table base_t( +id int primary key, +f32 float, +f64 double +); +insert into base_t values (1, 1.25, -2.5); +data branch create table src_t from base_t; +data branch create table dst_t from base_t; +insert into src_t values +(2, cast('NaN' as float), cast('NaN' as double)), +(3, cast('Inf' as float), cast('Inf' as double)), +(4, cast('-Inf' as float), cast('-Inf' as double)); +data branch diff src_t against dst_t output as diff_out; +➤ TABLE CREATED[12,0,0] 𝄀 +`br_float_special_values`.`diff_out` +select __mo_diff_flag, id, f32, f64 from diff_out order by id; +➤ __mo_diff_flag[12,-1,0] ¦ id[4,32,0] ¦ f32[7,24,0] ¦ f64[8,54,0] 𝄀 +INSERT ¦ 2 ¦ NaN ¦ NaN 𝄀 +INSERT ¦ 3 ¦ Infinity ¦ Infinity 𝄀 +INSERT ¦ 4 ¦ -Infinity ¦ -Infinity +data branch merge src_t into dst_t; +select id, f32, f64 from dst_t order by id; +➤ id[4,32,0] ¦ f32[7,24,0] ¦ f64[8,54,0] 𝄀 +1 ¦ 1.25 ¦ -2.5 𝄀 +2 ¦ NaN ¦ NaN 𝄀 +3 ¦ Infinity ¦ Infinity 𝄀 +4 ¦ -Infinity ¦ -Infinity +data branch create table pick_src from base_t; +data branch create table pick_dst from base_t; +insert into pick_src values +(2, cast('NaN' as float), cast('NaN' as double)), +(3, cast('Inf' as float), cast('Inf' as double)), +(4, cast('-Inf' as float), cast('-Inf' as double)); +data branch pick pick_src into pick_dst keys(2, 3, 4); +select id, f32, f64 from pick_dst order by id; +➤ id[4,32,0] ¦ f32[7,24,0] ¦ f64[8,54,0] 𝄀 +1 ¦ 1.25 ¦ -2.5 𝄀 +2 ¦ NaN ¦ NaN 𝄀 +3 ¦ Infinity ¦ Infinity 𝄀 +4 ¦ -Infinity ¦ -Infinity +create table no_pk_base(f double, note varchar(16)); +insert into no_pk_base values +(cast('NaN' as double), 'remove'), +(cast('Inf' as double), 'update'), +(cast('-Inf' as double), 'keep'); +data branch create table no_pk_src from no_pk_base; +data branch create table no_pk_dst from no_pk_base; +delete from no_pk_src where note = 'remove'; +update no_pk_src set note = 'updated' where note = 'update'; +data branch merge no_pk_src into no_pk_dst; +select f, note from no_pk_dst order by note; +➤ f[8,54,0] ¦ note[12,-1,0] 𝄀 +-Infinity ¦ keep 𝄀 +NaN ¦ remove 𝄀 +Infinity ¦ update 𝄀 +Infinity ¦ updated +create table portable_base(f32 float, f64 double, marker double, note varchar(16)); +insert into portable_base values +(cast('NaN' as float), cast('NaN' as double), null, 'remove'), +(cast('NaN' as float), cast('NaN' as double), cast('Inf' as double), 'update'), +(cast('Inf' as float), cast('-Inf' as double), null, 'keep'); +data branch create table portable_src from portable_base; +data branch create table portable_dst from portable_base; +delete from portable_src where note = 'remove'; +update portable_src set marker = cast('-Inf' as double), note = 'updated' where note = 'update'; +data branch diff portable_src against portable_dst output file '/tmp/'; +➤ FILE SAVED TO[12,0,0] ¦ HINT[12,0,0] 𝄀 +/tmp/diff_portable_src_portable_dst_20260803_132724_19045738-30c4-40e5-88e8-bca47f2f792f.sql ¦ DELETE FROM `br_float_special_values`.`portable_dst`, INSERT INTO `br_float_special_values`.`portable_dst` +delete from portable_dst +where serial(f32) = serial(cast('NaN' as float)) +and serial(f64) = serial(cast('NaN' as double)) +and marker is null and note = 'remove' +limit 1; +delete from portable_dst +where serial(f32) = serial(cast('NaN' as float)) +and serial(f64) = serial(cast('NaN' as double)) +and serial(marker) = serial(cast('Inf' as double)) and note = 'update' +limit 1; +insert into portable_dst values +(cast('NaN' as float), cast('NaN' as double), cast('-Inf' as double), 'updated'); +select f32, f64, marker, note from portable_dst order by note; +➤ f32[7,24,0] ¦ f64[8,54,0] ¦ marker[8,54,0] ¦ note[12,-1,0] 𝄀 +Infinity ¦ -Infinity ¦ null ¦ keep 𝄀 +NaN ¦ NaN ¦ -Infinity ¦ updated +create table real_float_base(k float primary key, note varchar(24)); +insert into real_float_base values +(cast('NaN' as float), 'update'), +(cast('Inf' as float), 'remove'), +(cast('-Inf' as float), 'keep'), +(1.5, 'finite'); +data branch create table real_float_src from real_float_base; +data branch create table real_float_dst from real_float_base; +update real_float_src set note = 'updated' where k != k; +delete from real_float_src where k = cast('Inf' as float); +data branch merge real_float_src into real_float_dst; +select k, note from real_float_dst order by note; +➤ k[7,24,0] ¦ note[12,-1,0] 𝄀 +1.5 ¦ finite 𝄀 +-Infinity ¦ keep 𝄀 +NaN ¦ updated +create table real_composite_base( +f32 float, +f64 double, +tag int, +note varchar(24), +primary key(f32, f64, tag) +); +insert into real_composite_base values +(cast('NaN' as float), 1.0, 1, 'update_f32'), +(1.0, cast('NaN' as double), 2, 'update_f64'), +(cast('Inf' as float), cast('-Inf' as double), 3, 'remove_inf'), +(cast('-Inf' as float), cast('Inf' as double), 4, 'keep_inf'), +(2.0, 3.0, 5, 'finite'); +data branch create table real_composite_src from real_composite_base; +data branch create table real_composite_dst from real_composite_base; +update real_composite_src set note = 'updated_f32' where f32 != f32; +update real_composite_src set note = 'updated_f64' where f64 != f64; +delete from real_composite_src where tag = 3; +data branch merge real_composite_src into real_composite_dst; +select f32, f64, tag, note from real_composite_dst order by tag; +➤ f32[7,24,0] ¦ f64[8,54,0] ¦ tag[4,32,0] ¦ note[12,-1,0] 𝄀 +NaN ¦ 1.0 ¦ 1 ¦ update_f32 𝄀 +1.0 ¦ NaN ¦ 2 ¦ update_f64 𝄀 +-Infinity ¦ Infinity ¦ 4 ¦ keep_inf 𝄀 +2.0 ¦ 3.0 ¦ 5 ¦ finite +data branch create table real_portable_src from real_composite_base; +data branch create table real_portable_dst from real_composite_base; +update real_portable_src set note = 'updated_f32' where f32 != f32; +update real_portable_src set note = 'updated_f64' where f64 != f64; +delete from real_portable_src where tag = 3; +data branch diff real_portable_src against real_portable_dst output file '/tmp/'; +➤ FILE SAVED TO[12,0,0] ¦ HINT[12,0,0] 𝄀 +/tmp/diff_real_portable_src_real_portable_dst_20260803_132724_280baa78-31f9-48d9-b8bd-1c002226ae5f.sql ¦ DELETE FROM `br_float_special_values`.`real_portable_dst`, INSERT INTO `br_float_special_values`.`real_portable_dst` +create table real_portable_delete_stage as +select f32 as branch_apply_key_0, f64 as branch_apply_key_1, tag as branch_apply_key_2 +from real_portable_dst where 1 = 0; +insert into real_portable_delete_stage values +(cast('NaN' as float), 1.0, 1), +(1.0, cast('NaN' as double), 2), +(cast('Inf' as float), cast('-Inf' as double), 3); +delete branch_apply_base +from real_portable_dst as branch_apply_base +join real_portable_delete_stage as branch_apply_stage on +serial(branch_apply_base.f32) = serial(branch_apply_stage.branch_apply_key_0) +and serial(branch_apply_base.f64) = serial(branch_apply_stage.branch_apply_key_1) +and branch_apply_base.tag = branch_apply_stage.branch_apply_key_2; +insert into real_portable_dst values +(cast('NaN' as float), 1.0, 1, 'updated_f32'), +(1.0, cast('NaN' as double), 2, 'updated_f64'); +drop table real_portable_delete_stage; +select f32, f64, tag, note from real_portable_dst order by tag; +➤ f32[7,24,0] ¦ f64[8,54,0] ¦ tag[4,32,0] ¦ note[12,-1,0] 𝄀 +NaN ¦ 1.0 ¦ 1 ¦ updated_f32 𝄀 +1.0 ¦ NaN ¦ 2 ¦ updated_f64 𝄀 +-Infinity ¦ Infinity ¦ 4 ¦ keep_inf 𝄀 +2.0 ¦ 3.0 ¦ 5 ¦ finite +create table bit_float_base(k float primary key, note varchar(24)); +insert into bit_float_base values(bit_cast(unhex('0000c07f') as float), 'nan0'); +insert into bit_float_base values(bit_cast(unhex('0100c07f') as float), 'nan1'); +insert into bit_float_base values(0.0, 'poszero'); +insert into bit_float_base values(bit_cast(unhex('00000080') as float), 'negzero'); +data branch create table bit_float_src from bit_float_base; +data branch create table bit_float_dst from bit_float_base; +update bit_float_src set note = 'nan1_updated' +where serial(k) = serial(bit_cast(unhex('0100c07f') as float)); +update bit_float_src set note = 'negzero_updated' +where serial(k) = serial(bit_cast(unhex('00000080') as float)); +data branch merge bit_float_src into bit_float_dst; +select note, hex(serial(k)) from bit_float_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan0 ¦ 20FFC00000 𝄀 +nan1_updated ¦ 20FFC00001 𝄀 +negzero_updated ¦ 207FFFFFFF 𝄀 +poszero ¦ 2080000000 +create table bit_double_base(k double primary key, note varchar(24)); +insert into bit_double_base values(bit_cast(unhex('000000000000f87f') as double), 'nan0'); +insert into bit_double_base values(bit_cast(unhex('010000000000f87f') as double), 'nan1'); +insert into bit_double_base values(0.0, 'poszero'); +insert into bit_double_base values(bit_cast(unhex('0000000000000080') as double), 'negzero'); +data branch create table bit_double_src from bit_double_base; +data branch create table bit_double_dst from bit_double_base; +update bit_double_src set note = 'nan1_updated' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +update bit_double_src set note = 'negzero_updated' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +data branch merge bit_double_src into bit_double_dst; +select note, hex(serial(k)) from bit_double_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan0 ¦ 21FFF8000000000000 𝄀 +nan1_updated ¦ 21FFF8000000000001 𝄀 +negzero_updated ¦ 217FFFFFFFFFFFFFFF 𝄀 +poszero ¦ 218000000000000000 +data branch create table bit_float_pick_dst from bit_float_base; +data branch pick bit_float_src into bit_float_pick_dst +keys(select k from bit_float_src where note in ('nan1_updated', 'negzero_updated')) +when conflict accept; +select note, hex(serial(k)) from bit_float_pick_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan0 ¦ 20FFC00000 𝄀 +nan1_updated ¦ 20FFC00001 𝄀 +negzero_updated ¦ 207FFFFFFF 𝄀 +poszero ¦ 2080000000 +data branch create table bit_double_pick_dst from bit_double_base; +data branch pick bit_double_src into bit_double_pick_dst +keys(select k from bit_double_src where note in ('nan1_updated', 'negzero_updated')) +when conflict accept; +select note, hex(serial(k)) from bit_double_pick_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan0 ¦ 21FFF8000000000000 𝄀 +nan1_updated ¦ 21FFF8000000000001 𝄀 +negzero_updated ¦ 217FFFFFFFFFFFFFFF 𝄀 +poszero ¦ 218000000000000000 +create table bit_composite_base( +f32 float, +f64 double, +tag int, +note varchar(32), +primary key(f32, f64, tag) +); +insert into bit_composite_base values +(bit_cast(unhex('0000c07f') as float), 2.0, 1, 'keep_f32_nan0'); +insert into bit_composite_base values +(bit_cast(unhex('0100c07f') as float), 2.0, 1, 'update_f32_nan1'); +insert into bit_composite_base values +(3.0, bit_cast(unhex('000000000000f87f') as double), 2, 'keep_f64_nan0'); +insert into bit_composite_base values +(3.0, bit_cast(unhex('010000000000f87f') as double), 2, 'update_f64_nan1'); +insert into bit_composite_base values(0.0, 0.0, 3, 'keep_poszero'); +insert into bit_composite_base values( +bit_cast(unhex('00000080') as float), +bit_cast(unhex('0000000000000080') as double), 3, 'update_negzero'); +data branch create table bit_composite_src from bit_composite_base; +data branch create table bit_composite_dst from bit_composite_base; +update bit_composite_src set note = 'updated_f32_nan1' +where serial(f32, f64, tag) = serial( +bit_cast(unhex('0100c07f') as float), 2.0, 1); +update bit_composite_src set note = 'updated_f64_nan1' +where serial(f32, f64, tag) = serial( +3.0, bit_cast(unhex('010000000000f87f') as double), 2); +update bit_composite_src set note = 'updated_negzero' +where serial(f32, f64, tag) = serial( +bit_cast(unhex('00000080') as float), +bit_cast(unhex('0000000000000080') as double), 3); +data branch merge bit_composite_src into bit_composite_dst; +select note, hex(serial(f32)), hex(serial(f64)), tag +from bit_composite_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(f32))[12,-1,0] ¦ hex(serial(f64))[12,-1,0] ¦ tag[4,32,0] 𝄀 +keep_f32_nan0 ¦ 20FFC00000 ¦ 21C000000000000000 ¦ 1 𝄀 +keep_f64_nan0 ¦ 20C0400000 ¦ 21FFF8000000000000 ¦ 2 𝄀 +keep_poszero ¦ 2080000000 ¦ 218000000000000000 ¦ 3 𝄀 +update_f32_nan1 ¦ 20FFC00001 ¦ 21C000000000000000 ¦ 1 𝄀 +update_f64_nan1 ¦ 20C0400000 ¦ 21FFF8000000000001 ¦ 2 𝄀 +update_negzero ¦ 207FFFFFFF ¦ 217FFFFFFFFFFFFFFF ¦ 3 +data branch create table bit_composite_pick_dst from bit_composite_base; +data branch pick bit_composite_src into bit_composite_pick_dst +keys(select f32, f64, tag from bit_composite_src +where note in ('updated_f32_nan1', 'updated_f64_nan1', 'updated_negzero')) +when conflict accept; +select note, hex(serial(f32)), hex(serial(f64)), tag +from bit_composite_pick_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(f32))[12,-1,0] ¦ hex(serial(f64))[12,-1,0] ¦ tag[4,32,0] 𝄀 +keep_f32_nan0 ¦ 20FFC00000 ¦ 21C000000000000000 ¦ 1 𝄀 +keep_f64_nan0 ¦ 20C0400000 ¦ 21FFF8000000000000 ¦ 2 𝄀 +keep_poszero ¦ 2080000000 ¦ 218000000000000000 ¦ 3 𝄀 +update_f32_nan1 ¦ 20FFC00001 ¦ 21C000000000000000 ¦ 1 𝄀 +update_f64_nan1 ¦ 20C0400000 ¦ 21FFF8000000000001 ¦ 2 𝄀 +update_negzero ¦ 207FFFFFFF ¦ 217FFFFFFFFFFFFFFF ¦ 3 +data branch create table bit_portable_src from bit_double_base; +data branch create table bit_portable_dst from bit_double_base; +update bit_portable_src set note = 'nan1_updated' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +update bit_portable_src set note = 'negzero_updated' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +data branch diff bit_portable_src against bit_portable_dst output file '/tmp/'; +➤ FILE SAVED TO[12,0,0] ¦ HINT[12,0,0] 𝄀 +/tmp/diff_bit_portable_src_bit_portable_dst_20260803_132725_357651cf-fbc1-4705-812b-17e7514ce0bf.sql ¦ DELETE FROM `br_float_special_values`.`bit_portable_dst`, INSERT INTO `br_float_special_values`.`bit_portable_dst` +update bit_portable_dst set note = 'nan1_updated' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)) limit 1; +update bit_portable_dst set note = 'negzero_updated' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)) limit 1; +select note, hex(serial(k)) from bit_portable_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan0 ¦ 21FFF8000000000000 𝄀 +nan1_updated ¦ 21FFF8000000000001 𝄀 +negzero_updated ¦ 217FFFFFFFFFFFFFFF 𝄀 +poszero ¦ 218000000000000000 +create table bit_missing_zero_base(k double primary key, note varchar(32)); +insert into bit_missing_zero_base values(0.0, 'poszero'); +insert into bit_missing_zero_base values( +bit_cast(unhex('0000000000000080') as double), 'negzero'); +create table bit_missing_nan_base(k double primary key, note varchar(32)); +insert into bit_missing_nan_base values( +bit_cast(unhex('000000000000f87f') as double), 'nan0'); +insert into bit_missing_nan_base values( +bit_cast(unhex('010000000000f87f') as double), 'nan1'); +data branch create table bit_zero_merge_src from bit_missing_zero_base; +data branch create table bit_zero_merge_dst from bit_missing_zero_base; +update bit_zero_merge_src set note = 'negzero_restored' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +delete from bit_zero_merge_dst +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +data branch merge bit_zero_merge_src into bit_zero_merge_dst when conflict accept; +select note, hex(serial(k)) from bit_zero_merge_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +negzero_restored ¦ 217FFFFFFFFFFFFFFF 𝄀 +poszero ¦ 218000000000000000 +data branch create table bit_nan_merge_src from bit_missing_nan_base; +data branch create table bit_nan_merge_dst from bit_missing_nan_base; +update bit_nan_merge_src set note = 'nan1_restored' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +delete from bit_nan_merge_dst +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +data branch merge bit_nan_merge_src into bit_nan_merge_dst when conflict accept; +select note, hex(serial(k)) from bit_nan_merge_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan0 ¦ 21FFF8000000000000 𝄀 +nan1_restored ¦ 21FFF8000000000001 +data branch create table bit_zero_pick_src from bit_missing_zero_base; +data branch create table bit_zero_pick_dst from bit_missing_zero_base; +update bit_zero_pick_src set note = 'negzero_restored' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +delete from bit_zero_pick_dst +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +data branch pick bit_zero_pick_src into bit_zero_pick_dst +keys(select k from bit_zero_pick_src where note = 'negzero_restored') +when conflict accept; +select note, hex(serial(k)) from bit_zero_pick_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +negzero_restored ¦ 217FFFFFFFFFFFFFFF 𝄀 +poszero ¦ 218000000000000000 +data branch create table bit_nan_pick_src from bit_missing_nan_base; +data branch create table bit_nan_pick_dst from bit_missing_nan_base; +update bit_nan_pick_src set note = 'nan1_restored' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +delete from bit_nan_pick_dst +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +data branch pick bit_nan_pick_src into bit_nan_pick_dst +keys(select k from bit_nan_pick_src where note = 'nan1_restored') +when conflict accept; +select note, hex(serial(k)) from bit_nan_pick_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan0 ¦ 21FFF8000000000000 𝄀 +nan1_restored ¦ 21FFF8000000000001 +data branch create table bit_zero_portable_src from bit_missing_zero_base; +data branch create table bit_zero_portable_dst from bit_missing_zero_base; +update bit_zero_portable_src set note = 'negzero_restored' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +delete from bit_zero_portable_dst +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +data branch diff bit_zero_portable_src against bit_zero_portable_dst output file '/tmp/'; +➤ FILE SAVED TO[12,0,0] ¦ HINT[12,0,0] 𝄀 +/tmp/diff_bit_zero_portable_src_bit_zero_portable_dst_20260803_151923_48b7be14-bb7d-48c4-a7c4-98a4aa54ee4d.sql ¦ DELETE FROM `br_float_special_values`.`bit_zero_portable_dst`, INSERT INTO `br_float_special_values`.`bit_zero_portable_dst` +insert into bit_zero_portable_dst(k, note) values +(bit_cast(unhex('0000000000000080') as double), 'negzero_restored'); +select note, hex(serial(k)) from bit_zero_portable_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +negzero_restored ¦ 217FFFFFFFFFFFFFFF 𝄀 +poszero ¦ 218000000000000000 +data branch create table bit_nan_portable_src from bit_missing_nan_base; +data branch create table bit_nan_portable_dst from bit_missing_nan_base; +update bit_nan_portable_src set note = 'nan1_restored' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +delete from bit_nan_portable_dst +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +data branch diff bit_nan_portable_src against bit_nan_portable_dst output file '/tmp/'; +➤ FILE SAVED TO[12,0,0] ¦ HINT[12,0,0] 𝄀 +/tmp/diff_bit_nan_portable_src_bit_nan_portable_dst_20260803_151923_cb57436e-4e70-41fc-980d-2eb94a016f39.sql ¦ DELETE FROM `br_float_special_values`.`bit_nan_portable_dst`, INSERT INTO `br_float_special_values`.`bit_nan_portable_dst` +insert into bit_nan_portable_dst(k, note) values +(bit_cast(unhex('010000000000f87f') as double), 'nan1_restored'); +select note, hex(serial(k)) from bit_nan_portable_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan0 ¦ 21FFF8000000000000 𝄀 +nan1_restored ¦ 21FFF8000000000001 +create table missing_float_base(k float primary key, note varchar(32)); +insert into missing_float_base values +(1.5, 'base'), (9.5, 'keep'); +data branch create table missing_float_src from missing_float_base; +data branch create table missing_float_fail_dst from missing_float_base; +data branch create table missing_float_skip_dst from missing_float_base; +data branch create table missing_float_accept_dst from missing_float_base; +update missing_float_src set note = 'source_updated' where k = 1.5; +delete from missing_float_fail_dst where k = 1.5; +delete from missing_float_skip_dst where k = 1.5; +delete from missing_float_accept_dst where k = 1.5; +data branch pick missing_float_src into missing_float_fail_dst +keys(select k from missing_float_src where note = 'source_updated') when conflict fail; +internal error: conflict: missing_float_src INSERT and missing_float_fail_dst INSERT on pk(1.5) with different values +select count(*) from missing_float_fail_dst; +➤ count(*)[-5,64,0] 𝄀 +1 +data branch pick missing_float_src into missing_float_skip_dst +keys(select k from missing_float_src where note = 'source_updated') when conflict skip; +select count(*) from missing_float_skip_dst; +➤ count(*)[-5,64,0] 𝄀 +1 +data branch pick missing_float_src into missing_float_accept_dst +keys(select k from missing_float_src where note = 'source_updated') when conflict accept; +select note, hex(serial(k)) from missing_float_accept_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +keep ¦ 20C1180000 𝄀 +source_updated ¦ 20BFC00000 +create table missing_double_base(k double primary key, note varchar(32)); +insert into missing_double_base values +(1.5, 'base'), (9.5, 'keep'); +data branch create table missing_double_src from missing_double_base; +data branch create table missing_double_fail_dst from missing_double_base; +data branch create table missing_double_skip_dst from missing_double_base; +data branch create table missing_double_accept_dst from missing_double_base; +update missing_double_src set note = 'source_updated' where k = 1.5; +delete from missing_double_fail_dst where k = 1.5; +delete from missing_double_skip_dst where k = 1.5; +delete from missing_double_accept_dst where k = 1.5; +data branch merge missing_double_src into missing_double_fail_dst when conflict fail; +internal error: conflict: missing_double_src DELETE and missing_double_fail_dst DELETE on pk(1.5) with different values +select count(*) from missing_double_fail_dst; +➤ count(*)[-5,64,0] 𝄀 +1 +data branch merge missing_double_src into missing_double_skip_dst when conflict skip; +select count(*) from missing_double_skip_dst; +➤ count(*)[-5,64,0] 𝄀 +1 +data branch merge missing_double_src into missing_double_accept_dst when conflict accept; +select note, hex(serial(k)) from missing_double_accept_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +keep ¦ 21C023000000000000 𝄀 +source_updated ¦ 21BFF8000000000000 +create table missing_composite_base( +f32 float, +tag int, +note varchar(32), +primary key(f32, tag) +); +insert into missing_composite_base values +(2.5, 7, 'base'), (9.5, 9, 'keep'); +data branch create table missing_composite_src from missing_composite_base; +data branch create table missing_composite_dst from missing_composite_base; +update missing_composite_src set note = 'source_updated' where tag = 7; +delete from missing_composite_dst where tag = 7; +data branch merge missing_composite_src into missing_composite_dst when conflict accept; +select note, hex(serial(f32)), tag from missing_composite_dst order by tag; +➤ note[12,-1,0] ¦ hex(serial(f32))[12,-1,0] ¦ tag[4,32,0] 𝄀 +source_updated ¦ 20C0200000 ¦ 7 𝄀 +keep ¦ 20C1180000 ¦ 9 +create table cross_zero_base(k float primary key, note varchar(32)); +insert into cross_zero_base values(0.0, 'poszero'); +insert into cross_zero_base values(bit_cast(unhex('00000080') as float), 'negzero'); +data branch create table cross_zero_src from cross_zero_base; +data branch create table cross_zero_fail_dst from cross_zero_base; +data branch create table cross_zero_skip_dst from cross_zero_base; +data branch create table cross_zero_accept_dst from cross_zero_base; +update cross_zero_src set note = 'negzero_cross_updated' +where serial(k) = serial(bit_cast(unhex('00000080') as float)); +delete from cross_zero_fail_dst +where serial(k) = serial(cast(0.0 as float)); +delete from cross_zero_skip_dst +where serial(k) = serial(cast(0.0 as float)); +delete from cross_zero_accept_dst +where serial(k) = serial(cast(0.0 as float)); +data branch merge cross_zero_src into cross_zero_fail_dst when conflict fail; +select note, hex(serial(k)) from cross_zero_fail_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +negzero_cross_updated ¦ 207FFFFFFF +data branch merge cross_zero_src into cross_zero_skip_dst when conflict skip; +select note, hex(serial(k)) from cross_zero_skip_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +negzero_cross_updated ¦ 207FFFFFFF +data branch merge cross_zero_src into cross_zero_accept_dst when conflict accept; +select note, hex(serial(k)) from cross_zero_accept_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +negzero_cross_updated ¦ 207FFFFFFF +data branch create table cross_nan_src from bit_missing_nan_base; +data branch create table cross_nan_fail_dst from bit_missing_nan_base; +data branch create table cross_nan_skip_dst from bit_missing_nan_base; +data branch create table cross_nan_accept_dst from bit_missing_nan_base; +update cross_nan_src set note = 'nan1_cross_updated' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +delete from cross_nan_fail_dst +where serial(k) = serial(bit_cast(unhex('000000000000f87f') as double)); +delete from cross_nan_skip_dst +where serial(k) = serial(bit_cast(unhex('000000000000f87f') as double)); +delete from cross_nan_accept_dst +where serial(k) = serial(bit_cast(unhex('000000000000f87f') as double)); +data branch pick cross_nan_src into cross_nan_fail_dst +keys(select k from cross_nan_src where note = 'nan1_cross_updated') when conflict fail; +select note, hex(serial(k)) from cross_nan_fail_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan1_cross_updated ¦ 21FFF8000000000001 +data branch pick cross_nan_src into cross_nan_skip_dst +keys(select k from cross_nan_src where note = 'nan1_cross_updated') when conflict skip; +select note, hex(serial(k)) from cross_nan_skip_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan1_cross_updated ¦ 21FFF8000000000001 +data branch pick cross_nan_src into cross_nan_accept_dst +keys(select k from cross_nan_src where note = 'nan1_cross_updated') when conflict accept; +select note, hex(serial(k)) from cross_nan_accept_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +nan1_cross_updated ¦ 21FFF8000000000001 +create table cross_composite_base( +k double, +tag int, +note varchar(32), +primary key(k, tag) +); +insert into cross_composite_base values(0.0, 7, 'poszero'); +insert into cross_composite_base values( +bit_cast(unhex('0000000000000080') as double), 7, 'negzero'); +data branch create table cross_composite_src from cross_composite_base; +data branch create table cross_composite_merge_dst from cross_composite_base; +data branch create table cross_composite_pick_dst from cross_composite_base; +update cross_composite_src set note = 'negzero_cross_updated' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +delete from cross_composite_merge_dst where serial(k) = serial(cast(0.0 as double)); +delete from cross_composite_pick_dst where serial(k) = serial(cast(0.0 as double)); +data branch merge cross_composite_src into cross_composite_merge_dst when conflict accept; +select note, hex(serial(k)), tag from cross_composite_merge_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] ¦ tag[4,32,0] 𝄀 +negzero_cross_updated ¦ 217FFFFFFFFFFFFFFF ¦ 7 +data branch pick cross_composite_src into cross_composite_pick_dst +keys(select k, tag from cross_composite_src where note = 'negzero_cross_updated') +when conflict accept; +select note, hex(serial(k)), tag from cross_composite_pick_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] ¦ tag[4,32,0] 𝄀 +negzero_cross_updated ¦ 217FFFFFFFFFFFFFFF ¦ 7 +create table missing_portable_base(k double primary key, note varchar(32)); +insert into missing_portable_base values +(1.5, 'base'), (9.5, 'keep'); +data branch create table missing_portable_src from missing_portable_base; +data branch create table missing_portable_dst from missing_portable_base; +update missing_portable_src set note = 'source_updated' where k = 1.5; +delete from missing_portable_dst where k = 1.5; +data branch diff missing_portable_src against missing_portable_dst output file '/tmp/'; +➤ FILE SAVED TO[12,0,0] ¦ HINT[12,0,0] 𝄀 +/tmp/diff_missing_portable_src_missing_portable_dst_20260803_132726_1435dfcc-41c4-4f79-80e5-55a523637d81.sql ¦ DELETE FROM `br_float_special_values`.`missing_portable_dst`, INSERT INTO `br_float_special_values`.`missing_portable_dst` +insert into missing_portable_dst(k, note) values +(cast(1.5 as double), 'source_updated'); +select note, hex(serial(k)) from missing_portable_dst order by note; +➤ note[12,-1,0] ¦ hex(serial(k))[12,-1,0] 𝄀 +keep ¦ 21C023000000000000 𝄀 +source_updated ¦ 21BFF8000000000000 +drop database br_float_special_values; diff --git a/test/distributed/cases/git4data/branch/edge/branch_float_special_values.sql b/test/distributed/cases/git4data/branch/edge/branch_float_special_values.sql new file mode 100644 index 0000000000000..b4d12008f5197 --- /dev/null +++ b/test/distributed/cases/git4data/branch/edge/branch_float_special_values.sql @@ -0,0 +1,481 @@ +-- DATA BRANCH SQL materialization must preserve every non-finite FLOAT/DOUBLE value. + +drop database if exists br_float_special_values; +create database br_float_special_values; +use br_float_special_values; + +create table base_t( + id int primary key, + f32 float, + f64 double +); +insert into base_t values (1, 1.25, -2.5); + +-- DIFF OUTPUT AS and MERGE share the row-to-SQL materialization path. +data branch create table src_t from base_t; +data branch create table dst_t from base_t; +insert into src_t values + (2, cast('NaN' as float), cast('NaN' as double)), + (3, cast('Inf' as float), cast('Inf' as double)), + (4, cast('-Inf' as float), cast('-Inf' as double)); + +data branch diff src_t against dst_t output as diff_out; +select __mo_diff_flag, id, f32, f64 from diff_out order by id; + +data branch merge src_t into dst_t; +select id, f32, f64 from dst_t order by id; + +-- PICK uses the same formatter through a separate SQL appender. +data branch create table pick_src from base_t; +data branch create table pick_dst from base_t; +insert into pick_src values + (2, cast('NaN' as float), cast('NaN' as double)), + (3, cast('Inf' as float), cast('Inf' as double)), + (4, cast('-Inf' as float), cast('-Inf' as double)); + +data branch pick pick_src into pick_dst keys(2, 3, 4); +select id, f32, f64 from pick_dst order by id; + +-- No-PK MERGE deletes by full row value before applying updates and deletes. +create table no_pk_base(f double, note varchar(16)); +insert into no_pk_base values + (cast('NaN' as double), 'remove'), + (cast('Inf' as double), 'update'), + (cast('-Inf' as double), 'keep'); +data branch create table no_pk_src from no_pk_base; +data branch create table no_pk_dst from no_pk_base; +delete from no_pk_src where note = 'remove'; +update no_pk_src set note = 'updated' where note = 'update'; + +data branch merge no_pk_src into no_pk_dst; +select f, note from no_pk_dst order by note; + +-- Portable SQL cannot use the hidden fake PK, so it deletes by full row. +-- Generate the portable script, then apply its exact NaN/NULL/infinity predicate +-- forms to the destination. Delete plus insert models the generated update and +-- proves that the old NaN row does not survive beside the new row. +create table portable_base(f32 float, f64 double, marker double, note varchar(16)); +insert into portable_base values + (cast('NaN' as float), cast('NaN' as double), null, 'remove'), + (cast('NaN' as float), cast('NaN' as double), cast('Inf' as double), 'update'), + (cast('Inf' as float), cast('-Inf' as double), null, 'keep'); +data branch create table portable_src from portable_base; +data branch create table portable_dst from portable_base; +delete from portable_src where note = 'remove'; +update portable_src set marker = cast('-Inf' as double), note = 'updated' where note = 'update'; + +-- @ignore:0,1 +data branch diff portable_src against portable_dst output file '/tmp/'; + +delete from portable_dst +where serial(f32) = serial(cast('NaN' as float)) + and serial(f64) = serial(cast('NaN' as double)) + and marker is null and note = 'remove' +limit 1; +delete from portable_dst +where serial(f32) = serial(cast('NaN' as float)) + and serial(f64) = serial(cast('NaN' as double)) + and serial(marker) = serial(cast('Inf' as double)) and note = 'update' +limit 1; +insert into portable_dst values + (cast('NaN' as float), cast('NaN' as double), cast('-Inf' as double), 'updated'); + +select f32, f64, marker, note from portable_dst order by note; + +-- A real FLOAT primary key must use exact bit identity in both the LCA probe +-- and the staged delete. Infinity and finite keys retain ordinary equality. +create table real_float_base(k float primary key, note varchar(24)); +insert into real_float_base values + (cast('NaN' as float), 'update'), + (cast('Inf' as float), 'remove'), + (cast('-Inf' as float), 'keep'), + (1.5, 'finite'); +data branch create table real_float_src from real_float_base; +data branch create table real_float_dst from real_float_base; +update real_float_src set note = 'updated' where k != k; +delete from real_float_src where k = cast('Inf' as float); + +data branch merge real_float_src into real_float_dst; +select k, note from real_float_dst order by note; + +-- Composite real keys need the same rule independently for every FLOAT and +-- DOUBLE component. These rows place NaN in each component and retain +-- finite/+Inf/-Inf controls. +create table real_composite_base( + f32 float, + f64 double, + tag int, + note varchar(24), + primary key(f32, f64, tag) +); +insert into real_composite_base values + (cast('NaN' as float), 1.0, 1, 'update_f32'), + (1.0, cast('NaN' as double), 2, 'update_f64'), + (cast('Inf' as float), cast('-Inf' as double), 3, 'remove_inf'), + (cast('-Inf' as float), cast('Inf' as double), 4, 'keep_inf'), + (2.0, 3.0, 5, 'finite'); +data branch create table real_composite_src from real_composite_base; +data branch create table real_composite_dst from real_composite_base; +update real_composite_src set note = 'updated_f32' where f32 != f32; +update real_composite_src set note = 'updated_f64' where f64 != f64; +delete from real_composite_src where tag = 3; + +data branch merge real_composite_src into real_composite_dst; +select f32, f64, tag, note from real_composite_dst order by tag; + +-- Generate the public portable-SQL path for the same composite real key. The +-- deterministic statements below apply the generator's exact staged-delete +-- predicate shape so the expected final table remains an executable oracle. +data branch create table real_portable_src from real_composite_base; +data branch create table real_portable_dst from real_composite_base; +update real_portable_src set note = 'updated_f32' where f32 != f32; +update real_portable_src set note = 'updated_f64' where f64 != f64; +delete from real_portable_src where tag = 3; + +-- @ignore:0,1 +data branch diff real_portable_src against real_portable_dst output file '/tmp/'; + +create table real_portable_delete_stage as +select f32 as branch_apply_key_0, f64 as branch_apply_key_1, tag as branch_apply_key_2 +from real_portable_dst where 1 = 0; +insert into real_portable_delete_stage values + (cast('NaN' as float), 1.0, 1), + (1.0, cast('NaN' as double), 2), + (cast('Inf' as float), cast('-Inf' as double), 3); +delete branch_apply_base +from real_portable_dst as branch_apply_base +join real_portable_delete_stage as branch_apply_stage on + serial(branch_apply_base.f32) = serial(branch_apply_stage.branch_apply_key_0) + and serial(branch_apply_base.f64) = serial(branch_apply_stage.branch_apply_key_1) + and branch_apply_base.tag = branch_apply_stage.branch_apply_key_2; +insert into real_portable_dst values + (cast('NaN' as float), 1.0, 1, 'updated_f32'), + (1.0, cast('NaN' as double), 2, 'updated_f64'); +drop table real_portable_delete_stage; + +select f32, f64, tag, note from real_portable_dst order by tag; + +-- MatrixOne primary keys preserve FLOAT/DOUBLE bits. Scalar equality cannot +-- distinguish NaN payloads or signed zero, so exercise each representation +-- through the public MERGE path and use serial() as an independent bit oracle. +create table bit_float_base(k float primary key, note varchar(24)); +insert into bit_float_base values(bit_cast(unhex('0000c07f') as float), 'nan0'); +insert into bit_float_base values(bit_cast(unhex('0100c07f') as float), 'nan1'); +insert into bit_float_base values(0.0, 'poszero'); +insert into bit_float_base values(bit_cast(unhex('00000080') as float), 'negzero'); +data branch create table bit_float_src from bit_float_base; +data branch create table bit_float_dst from bit_float_base; +update bit_float_src set note = 'nan1_updated' +where serial(k) = serial(bit_cast(unhex('0100c07f') as float)); +update bit_float_src set note = 'negzero_updated' +where serial(k) = serial(bit_cast(unhex('00000080') as float)); +data branch merge bit_float_src into bit_float_dst; +select note, hex(serial(k)) from bit_float_dst order by note; + +create table bit_double_base(k double primary key, note varchar(24)); +insert into bit_double_base values(bit_cast(unhex('000000000000f87f') as double), 'nan0'); +insert into bit_double_base values(bit_cast(unhex('010000000000f87f') as double), 'nan1'); +insert into bit_double_base values(0.0, 'poszero'); +insert into bit_double_base values(bit_cast(unhex('0000000000000080') as double), 'negzero'); +data branch create table bit_double_src from bit_double_base; +data branch create table bit_double_dst from bit_double_base; +update bit_double_src set note = 'nan1_updated' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +update bit_double_src set note = 'negzero_updated' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +data branch merge bit_double_src into bit_double_dst; +select note, hex(serial(k)) from bit_double_dst order by note; + +-- PICK applies accepted same-key changes through the same exact-key upsert +-- path as MERGE. The destination must retain the unpicked NaN payload and +-- positive-zero rows rather than deleting/reinserting equivalent scalar keys. +data branch create table bit_float_pick_dst from bit_float_base; +data branch pick bit_float_src into bit_float_pick_dst +keys(select k from bit_float_src where note in ('nan1_updated', 'negzero_updated')) +when conflict accept; +select note, hex(serial(k)) from bit_float_pick_dst order by note; + +data branch create table bit_double_pick_dst from bit_double_base; +data branch pick bit_double_src into bit_double_pick_dst +keys(select k from bit_double_src where note in ('nan1_updated', 'negzero_updated')) +when conflict accept; +select note, hex(serial(k)) from bit_double_pick_dst order by note; + +-- Composite keys apply exact identity independently to both float widths; +-- paired rows differ only in the selected float representation. +create table bit_composite_base( + f32 float, + f64 double, + tag int, + note varchar(32), + primary key(f32, f64, tag) +); +insert into bit_composite_base values + (bit_cast(unhex('0000c07f') as float), 2.0, 1, 'keep_f32_nan0'); +insert into bit_composite_base values + (bit_cast(unhex('0100c07f') as float), 2.0, 1, 'update_f32_nan1'); +insert into bit_composite_base values + (3.0, bit_cast(unhex('000000000000f87f') as double), 2, 'keep_f64_nan0'); +insert into bit_composite_base values + (3.0, bit_cast(unhex('010000000000f87f') as double), 2, 'update_f64_nan1'); +insert into bit_composite_base values(0.0, 0.0, 3, 'keep_poszero'); +insert into bit_composite_base values( + bit_cast(unhex('00000080') as float), + bit_cast(unhex('0000000000000080') as double), 3, 'update_negzero'); +data branch create table bit_composite_src from bit_composite_base; +data branch create table bit_composite_dst from bit_composite_base; +update bit_composite_src set note = 'updated_f32_nan1' +where serial(f32, f64, tag) = serial( + bit_cast(unhex('0100c07f') as float), 2.0, 1); +update bit_composite_src set note = 'updated_f64_nan1' +where serial(f32, f64, tag) = serial( + 3.0, bit_cast(unhex('010000000000f87f') as double), 2); +update bit_composite_src set note = 'updated_negzero' +where serial(f32, f64, tag) = serial( + bit_cast(unhex('00000080') as float), + bit_cast(unhex('0000000000000080') as double), 3); +data branch merge bit_composite_src into bit_composite_dst; +select note, hex(serial(f32)), hex(serial(f64)), tag +from bit_composite_dst order by note; + +data branch create table bit_composite_pick_dst from bit_composite_base; +data branch pick bit_composite_src into bit_composite_pick_dst +keys(select f32, f64, tag from bit_composite_src + where note in ('updated_f32_nan1', 'updated_f64_nan1', 'updated_negzero')) +when conflict accept; +select note, hex(serial(f32)), hex(serial(f64)), tag +from bit_composite_pick_dst order by note; + +-- Portable real-key SQL upserts rows with exact bit-preserving literals. +data branch create table bit_portable_src from bit_double_base; +data branch create table bit_portable_dst from bit_double_base; +update bit_portable_src set note = 'nan1_updated' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +update bit_portable_src set note = 'negzero_updated' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +-- @ignore:0,1 +data branch diff bit_portable_src against bit_portable_dst output file '/tmp/'; +update bit_portable_dst set note = 'nan1_updated' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)) limit 1; +update bit_portable_dst set note = 'negzero_updated' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)) limit 1; +select note, hex(serial(k)) from bit_portable_dst order by note; + +-- A missing updated key can still have a bit-distinct scalar peer in snapshot +-- storage. MERGE, PICK, and portable SQL must materialize that peer class and +-- then restore each accepted row with an independent INSERT ... VALUES. +create table bit_missing_zero_base(k double primary key, note varchar(32)); +insert into bit_missing_zero_base values(0.0, 'poszero'); +insert into bit_missing_zero_base values( + bit_cast(unhex('0000000000000080') as double), 'negzero'); +create table bit_missing_nan_base(k double primary key, note varchar(32)); +insert into bit_missing_nan_base values( + bit_cast(unhex('000000000000f87f') as double), 'nan0'); +insert into bit_missing_nan_base values( + bit_cast(unhex('010000000000f87f') as double), 'nan1'); + +data branch create table bit_zero_merge_src from bit_missing_zero_base; +data branch create table bit_zero_merge_dst from bit_missing_zero_base; +update bit_zero_merge_src set note = 'negzero_restored' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +delete from bit_zero_merge_dst +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +data branch merge bit_zero_merge_src into bit_zero_merge_dst when conflict accept; +select note, hex(serial(k)) from bit_zero_merge_dst order by note; + +data branch create table bit_nan_merge_src from bit_missing_nan_base; +data branch create table bit_nan_merge_dst from bit_missing_nan_base; +update bit_nan_merge_src set note = 'nan1_restored' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +delete from bit_nan_merge_dst +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +data branch merge bit_nan_merge_src into bit_nan_merge_dst when conflict accept; +select note, hex(serial(k)) from bit_nan_merge_dst order by note; + +data branch create table bit_zero_pick_src from bit_missing_zero_base; +data branch create table bit_zero_pick_dst from bit_missing_zero_base; +update bit_zero_pick_src set note = 'negzero_restored' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +delete from bit_zero_pick_dst +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +data branch pick bit_zero_pick_src into bit_zero_pick_dst +keys(select k from bit_zero_pick_src where note = 'negzero_restored') +when conflict accept; +select note, hex(serial(k)) from bit_zero_pick_dst order by note; + +data branch create table bit_nan_pick_src from bit_missing_nan_base; +data branch create table bit_nan_pick_dst from bit_missing_nan_base; +update bit_nan_pick_src set note = 'nan1_restored' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +delete from bit_nan_pick_dst +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +data branch pick bit_nan_pick_src into bit_nan_pick_dst +keys(select k from bit_nan_pick_src where note = 'nan1_restored') +when conflict accept; +select note, hex(serial(k)) from bit_nan_pick_dst order by note; + +data branch create table bit_zero_portable_src from bit_missing_zero_base; +data branch create table bit_zero_portable_dst from bit_missing_zero_base; +update bit_zero_portable_src set note = 'negzero_restored' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +delete from bit_zero_portable_dst +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +-- @ignore:0,1 +data branch diff bit_zero_portable_src against bit_zero_portable_dst output file '/tmp/'; +insert into bit_zero_portable_dst(k, note) values + (bit_cast(unhex('0000000000000080') as double), 'negzero_restored'); +select note, hex(serial(k)) from bit_zero_portable_dst order by note; + +data branch create table bit_nan_portable_src from bit_missing_nan_base; +data branch create table bit_nan_portable_dst from bit_missing_nan_base; +update bit_nan_portable_src set note = 'nan1_restored' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +delete from bit_nan_portable_dst +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +-- @ignore:0,1 +data branch diff bit_nan_portable_src against bit_nan_portable_dst output file '/tmp/'; +insert into bit_nan_portable_dst(k, note) values + (bit_cast(unhex('010000000000f87f') as double), 'nan1_restored'); +select note, hex(serial(k)) from bit_nan_portable_dst order by note; + +-- A source update conflicts with an independent destination delete. ACCEPT +-- must restore the source row; FAIL and SKIP must retain the destination delete. +create table missing_float_base(k float primary key, note varchar(32)); +insert into missing_float_base values + (1.5, 'base'), (9.5, 'keep'); +data branch create table missing_float_src from missing_float_base; +data branch create table missing_float_fail_dst from missing_float_base; +data branch create table missing_float_skip_dst from missing_float_base; +data branch create table missing_float_accept_dst from missing_float_base; +update missing_float_src set note = 'source_updated' where k = 1.5; +delete from missing_float_fail_dst where k = 1.5; +delete from missing_float_skip_dst where k = 1.5; +delete from missing_float_accept_dst where k = 1.5; +data branch pick missing_float_src into missing_float_fail_dst + keys(select k from missing_float_src where note = 'source_updated') when conflict fail; +select count(*) from missing_float_fail_dst; +data branch pick missing_float_src into missing_float_skip_dst + keys(select k from missing_float_src where note = 'source_updated') when conflict skip; +select count(*) from missing_float_skip_dst; +data branch pick missing_float_src into missing_float_accept_dst + keys(select k from missing_float_src where note = 'source_updated') when conflict accept; +select note, hex(serial(k)) from missing_float_accept_dst order by note; + +create table missing_double_base(k double primary key, note varchar(32)); +insert into missing_double_base values + (1.5, 'base'), (9.5, 'keep'); +data branch create table missing_double_src from missing_double_base; +data branch create table missing_double_fail_dst from missing_double_base; +data branch create table missing_double_skip_dst from missing_double_base; +data branch create table missing_double_accept_dst from missing_double_base; +update missing_double_src set note = 'source_updated' where k = 1.5; +delete from missing_double_fail_dst where k = 1.5; +delete from missing_double_skip_dst where k = 1.5; +delete from missing_double_accept_dst where k = 1.5; +data branch merge missing_double_src into missing_double_fail_dst when conflict fail; +select count(*) from missing_double_fail_dst; +data branch merge missing_double_src into missing_double_skip_dst when conflict skip; +select count(*) from missing_double_skip_dst; +data branch merge missing_double_src into missing_double_accept_dst when conflict accept; +select note, hex(serial(k)) from missing_double_accept_dst order by note; + +create table missing_composite_base( + f32 float, + tag int, + note varchar(32), + primary key(f32, tag) +); +insert into missing_composite_base values + (2.5, 7, 'base'), (9.5, 9, 'keep'); +data branch create table missing_composite_src from missing_composite_base; +data branch create table missing_composite_dst from missing_composite_base; +update missing_composite_src set note = 'source_updated' where tag = 7; +delete from missing_composite_dst where tag = 7; +data branch merge missing_composite_src into missing_composite_dst when conflict accept; +select note, hex(serial(f32)), tag from missing_composite_dst order by tag; + +-- Crossed changes on bit-distinct keys are independent, not conflicts. The +-- conflict join must not pair +0 with -0 or different NaN payloads under any +-- policy. +create table cross_zero_base(k float primary key, note varchar(32)); +insert into cross_zero_base values(0.0, 'poszero'); +insert into cross_zero_base values(bit_cast(unhex('00000080') as float), 'negzero'); +data branch create table cross_zero_src from cross_zero_base; +data branch create table cross_zero_fail_dst from cross_zero_base; +data branch create table cross_zero_skip_dst from cross_zero_base; +data branch create table cross_zero_accept_dst from cross_zero_base; +update cross_zero_src set note = 'negzero_cross_updated' +where serial(k) = serial(bit_cast(unhex('00000080') as float)); +delete from cross_zero_fail_dst +where serial(k) = serial(cast(0.0 as float)); +delete from cross_zero_skip_dst +where serial(k) = serial(cast(0.0 as float)); +delete from cross_zero_accept_dst +where serial(k) = serial(cast(0.0 as float)); +data branch merge cross_zero_src into cross_zero_fail_dst when conflict fail; +select note, hex(serial(k)) from cross_zero_fail_dst order by note; +data branch merge cross_zero_src into cross_zero_skip_dst when conflict skip; +select note, hex(serial(k)) from cross_zero_skip_dst order by note; +data branch merge cross_zero_src into cross_zero_accept_dst when conflict accept; +select note, hex(serial(k)) from cross_zero_accept_dst order by note; + +data branch create table cross_nan_src from bit_missing_nan_base; +data branch create table cross_nan_fail_dst from bit_missing_nan_base; +data branch create table cross_nan_skip_dst from bit_missing_nan_base; +data branch create table cross_nan_accept_dst from bit_missing_nan_base; +update cross_nan_src set note = 'nan1_cross_updated' +where serial(k) = serial(bit_cast(unhex('010000000000f87f') as double)); +delete from cross_nan_fail_dst +where serial(k) = serial(bit_cast(unhex('000000000000f87f') as double)); +delete from cross_nan_skip_dst +where serial(k) = serial(bit_cast(unhex('000000000000f87f') as double)); +delete from cross_nan_accept_dst +where serial(k) = serial(bit_cast(unhex('000000000000f87f') as double)); +data branch pick cross_nan_src into cross_nan_fail_dst +keys(select k from cross_nan_src where note = 'nan1_cross_updated') when conflict fail; +select note, hex(serial(k)) from cross_nan_fail_dst order by note; +data branch pick cross_nan_src into cross_nan_skip_dst +keys(select k from cross_nan_src where note = 'nan1_cross_updated') when conflict skip; +select note, hex(serial(k)) from cross_nan_skip_dst order by note; +data branch pick cross_nan_src into cross_nan_accept_dst +keys(select k from cross_nan_src where note = 'nan1_cross_updated') when conflict accept; +select note, hex(serial(k)) from cross_nan_accept_dst order by note; + +create table cross_composite_base( + k double, + tag int, + note varchar(32), + primary key(k, tag) +); +insert into cross_composite_base values(0.0, 7, 'poszero'); +insert into cross_composite_base values( + bit_cast(unhex('0000000000000080') as double), 7, 'negzero'); +data branch create table cross_composite_src from cross_composite_base; +data branch create table cross_composite_merge_dst from cross_composite_base; +data branch create table cross_composite_pick_dst from cross_composite_base; +update cross_composite_src set note = 'negzero_cross_updated' +where serial(k) = serial(bit_cast(unhex('0000000000000080') as double)); +delete from cross_composite_merge_dst where serial(k) = serial(cast(0.0 as double)); +delete from cross_composite_pick_dst where serial(k) = serial(cast(0.0 as double)); +data branch merge cross_composite_src into cross_composite_merge_dst when conflict accept; +select note, hex(serial(k)), tag from cross_composite_merge_dst order by note; +data branch pick cross_composite_src into cross_composite_pick_dst +keys(select k, tag from cross_composite_src where note = 'negzero_cross_updated') +when conflict accept; +select note, hex(serial(k)), tag from cross_composite_pick_dst order by note; + +-- Round-trip the portable statement shape for a destination-delete conflict. +create table missing_portable_base(k double primary key, note varchar(32)); +insert into missing_portable_base values + (1.5, 'base'), (9.5, 'keep'); +data branch create table missing_portable_src from missing_portable_base; +data branch create table missing_portable_dst from missing_portable_base; +update missing_portable_src set note = 'source_updated' where k = 1.5; +delete from missing_portable_dst where k = 1.5; +-- @ignore:0,1 +data branch diff missing_portable_src against missing_portable_dst output file '/tmp/'; +insert into missing_portable_dst(k, note) values + (cast(1.5 as double), 'source_updated'); +select note, hex(serial(k)) from missing_portable_dst order by note; + +drop database br_float_special_values;