fix(data-branch): improve diff correctness, memory control, and output summary support

pkg/common/rscthrottler/resource_throttler.go

@@ -363,3 +363,19 @@ func AcquirePolicyForCNFlushS3(
 
 	return defaultAcquirePolicy(throttler, ask)
 }
+
+func AcquirePolicyForDataBranch(
+	throttler *memThrottler,
+	ask int64,
+) (int64, bool) {
+
+	total := int64(throttler.actualTotalMemory.Load())
+	if total > 0 && throttler.limitRate > 0 {
+		used := throttler.rss.Load() + ask
+		if float64(used) > float64(total)*throttler.limitRate {
+			return 0, false
+		}
+	}
+
+	return defaultAcquirePolicy(throttler, ask)
+}

pkg/common/rscthrottler/resource_throttler_test.go

@@ -130,3 +130,42 @@ func BenchmarkThrottler(b *testing.B) {
 		throttler.Available()
 	}
 }
+
+func TestAcquirePolicyForDataBranch(t *testing.T) {
+	t.Run("deny when projected usage exceeds rate limit", func(t *testing.T) {
+		throttler := &memThrottler{limitRate: 0.80}
+		throttler.actualTotalMemory.Store(100)
+		throttler.limit.Store(80)
+		throttler.rss.Store(70)
+		throttler.reserved.Store(5)
+
+		left, ok := AcquirePolicyForDataBranch(throttler, 11)
+		require.False(t, ok)
+		require.Equal(t, int64(0), left)
+		require.Equal(t, int64(5), throttler.reserved.Load())
+	})
+
+	t.Run("allow at boundary and reserve memory", func(t *testing.T) {
+		throttler := &memThrottler{limitRate: 0.80}
+		throttler.actualTotalMemory.Store(100)
+		throttler.limit.Store(80)
+		throttler.rss.Store(70)
+
+		left, ok := AcquirePolicyForDataBranch(throttler, 10)
+		require.True(t, ok)
+		require.Equal(t, int64(20), left)
+		require.Equal(t, int64(10), throttler.reserved.Load())
+	})
+
+	t.Run("fallback to default policy when rate check is disabled", func(t *testing.T) {
+		throttler := &memThrottler{limitRate: 0}
+		throttler.actualTotalMemory.Store(100)
+		throttler.limit.Store(80)
+		throttler.rss.Store(20)
+
+		left, ok := AcquirePolicyForDataBranch(throttler, 10)
+		require.True(t, ok)
+		require.Equal(t, int64(70), left)
+		require.Equal(t, int64(10), throttler.reserved.Load())
+	})
+}

pkg/frontend/clone.go

@@ -599,11 +599,18 @@ func updateBranchMetaTable(
 		return nil
 	}
 
-	if _, srcTblDef, err = ses.GetTxnCompileCtx().Resolve(
-		receipt.srcDb, receipt.srcTbl, receipt.snapshot,
-	); err != nil {
+	srcCtx := defines.AttachAccountId(ctx, receipt.srcAccount)
+	tcc := ses.GetTxnCompileCtx()
+	origCtx := tcc.GetContext()
+	tcc.SetContext(srcCtx)
+	defer tcc.SetContext(origCtx)
+
+	if _, srcTblDef, err = tcc.Resolve(receipt.srcDb, receipt.srcTbl, nil); err != nil {
 		return err
 	}
+	if srcTblDef == nil {
+		return moerr.NewNoSuchTable(srcCtx, receipt.srcDb, receipt.srcTbl)
+	}
 
 	dstCtx := defines.AttachAccountId(ctx, receipt.toAccount)