Agent/catalog incremental discovery

pkg/catalog/types.go

@@ -55,6 +55,13 @@ const (
 	TableTailAttrAborted     = objectio.TombstoneAttr_Abort_Attr
 	TableTailAttrPKVal       = objectio.TombstoneAttr_PK_Attr
 
+	// TableChanges metadata columns are reserved because table_changes exposes
+	// them in the same row shape as source-table columns.
+	TableChangesAttrChangeType    = "change_type"
+	TableChangesAttrCommitTS      = "commit_ts"
+	TableChangesAttrTableID       = "table_id"
+	TableChangesAttrSchemaVersion = "schema_version"
+
 	MOAccountTable = "mo_account"
 	// MOVersionTable mo version table. This table records information about the
 	// versions of the MO cluster that have been upgraded. In other words, you can
@@ -125,6 +132,18 @@ const (
 	System_Account = uint32(0)
 )
 
+func IsTableChangesMetadataColumn(name string) bool {
+	switch strings.ToLower(name) {
+	case TableChangesAttrChangeType,
+		TableChangesAttrCommitTS,
+		TableChangesAttrTableID,
+		TableChangesAttrSchemaVersion:
+		return true
+	default:
+		return false
+	}
+}
+
 const (
 	MO_COMMENT_NO_DEL_HINT = "[mo_no_del_hint]"
 )

pkg/compress/compress_test.go

@@ -41,3 +41,9 @@ func TestLz4(t *testing.T) {
 	}
 	fmt.Printf("dat: %v\n", data)
 }
+
+func TestTypeString(t *testing.T) {
+	if got := T(Lz4Chunked).String(); got != "LZ4Chunked" {
+		t.Fatalf("unexpected chunked compression name %q", got)
+	}
+}

pkg/compress/types.go

@@ -19,6 +19,7 @@ import "fmt"
 const (
 	None = iota
 	Lz4
+	Lz4Chunked
 )
 
 type T uint8
@@ -29,6 +30,8 @@ func (t T) String() string {
 		return "None"
 	case Lz4:
 		return "LZ4"
+	case Lz4Chunked:
+		return "LZ4Chunked"
 	}
 	return fmt.Sprintf("unexpected compress type: %d", t)
 }

pkg/container/vector/vector.go

@@ -5310,6 +5310,7 @@ func unionT[T int32 | int64](v, w *Vector, sels []T, mp *mpool.MPool) error {
 }
 
 func (v *Vector) UnionBatch(w *Vector, offset int64, cnt int, flags []uint8, mp *mpool.MPool) error {
+	areaWasDisjoint := v.areaDisjoint
 	if v.typ.IsVarlen() {
 		v.areaDisjoint = false
 	}
@@ -5443,6 +5444,11 @@ func (v *Vector) UnionBatch(w *Vector, offset int64, cnt int, flags []uint8, mp
 			if err := v.propagatePrepareParamKindsBatch(w, oldLen, offset, cnt, flags, mp); err != nil {
 				return err
 			}
+			// The fast path copies the source area once into a non-overlapping
+			// destination range and only rebases descriptors. Therefore it
+			// preserves the proof exactly when both existing destination and
+			// source were already proven disjoint.
+			v.areaDisjoint = areaWasDisjoint && w.VarlenaAreaIsDisjoint()
 			return nil
 		}
 

pkg/container/vector/vector_test.go

@@ -3868,6 +3868,12 @@ func TestVarlenaAreaDisjointLifecycle(t *testing.T) {
 	for range 2 {
 		require.NoError(t, AppendBytes(flat, payload, false, mp))
 	}
+	copied := NewVec(typ)
+	require.NoError(t, copied.UnionBatch(flat, 0, flat.Length(), nil, mp))
+	require.True(t, copied.VarlenaAreaIsDisjoint(),
+		"full in-order copy preserves independent payload ranges")
+	copied.Free(mp)
+
 	flat.Shrink([]int64{0, 0}, false)
 	require.False(t, flat.VarlenaAreaIsDisjoint(),
 		"selection can duplicate a descriptor")

pkg/defines/const.go

@@ -51,7 +51,8 @@ const (
 	MORPCVersion13     int64 = 13 // lossless v2 prefix-index metadata
 	MORPCVersion14     int64 = 14 // utf8mb4 text MIN/MAX collation semantics
 	MORPCVersion15     int64 = 15 // CHECK metadata in rename-column alter requests
-	MORPCLatestVersion       = MORPCVersion15
+	MORPCVersion16     int64 = 16 // chunked persisted column extents
+	MORPCLatestVersion       = MORPCVersion16
 )
 
 // DefaultLockWaitTimeoutSeconds is shared by the frontend default and by

pkg/frontend/authenticate.go

@@ -6955,9 +6955,11 @@ func extractPrivilegeTipsFromPlan(p *plan2.Plan) privilegeTipsArray {
 		}
 
 		for nodeID, node := range q.Nodes {
-			if node.NodeType == plan.Node_TABLE_SCAN {
-				if _, ok := insertDedupScans[int32(nodeID)]; ok {
-					continue
+			if isPrivilegeBearingTableScan(node) {
+				if node.NodeType == plan.Node_TABLE_SCAN {
+					if _, ok := insertDedupScans[int32(nodeID)]; ok {
+						continue
+					}
 				}
 				if node.ObjRef != nil {
 					if node.TableDef != nil && node.TableDef.TableType == catalog.SystemClusterRel {
@@ -7234,6 +7236,17 @@ func extractPrivilegeTipsFromPlan(p *plan2.Plan) privilegeTipsArray {
 	return pts
 }
 
+func isPrivilegeBearingTableScan(node *plan.Node) bool {
+	if node == nil {
+		return false
+	}
+	if node.NodeType == plan.Node_TABLE_SCAN {
+		return true
+	}
+	return node.NodeType == plan.Node_FUNCTION_SCAN &&
+		node.GetTableDef().GetTblFunc().GetName() == "table_changes"
+}
+
 func addReplaceDeletePrivilegeTips(arr privilegeTipsArray, p *plan2.Plan) privilegeTipsArray {
 	if p.GetQuery() == nil {
 		return arr

pkg/frontend/authenticate_test.go

@@ -6031,6 +6031,54 @@ func Test_extractPrivilegeTipsFromPlan_Subscription(t *testing.T) {
 	assert.Equal(t, "t1", arr[0].tableName)
 }
 
+func TestExtractPrivilegeTipsFromTableChanges(t *testing.T) {
+	tests := []struct {
+		name           string
+		databaseName   string
+		tableName      string
+		isClusterTable bool
+	}{
+		{
+			name:         "ordinary table",
+			databaseName: "db1",
+			tableName:    "secret",
+		},
+		{
+			name:           "cluster table",
+			databaseName:   moCatalog,
+			tableName:      "tenant_secret",
+			isClusterTable: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			p := &plan2.Plan{Plan: &plan2.Plan_Query{Query: &plan2.Query{
+				StmtType: plan.Query_SELECT,
+				Nodes: []*plan2.Node{{
+					NodeType: plan.Node_FUNCTION_SCAN,
+					ObjRef: &plan2.ObjectRef{
+						SchemaName: tt.databaseName,
+						ObjName:    tt.tableName,
+					},
+					TableDef: &plan2.TableDef{
+						TableType: "func_table",
+						TblFunc:   &plan.TableFunction{Name: "table_changes"},
+					},
+				}},
+			}}}
+
+			arr := extractPrivilegeTipsFromPlan(p)
+			require.Len(t, arr, 1)
+			assert.Equal(t, PrivilegeTypeSelect, arr[0].typ)
+			assert.Equal(t, tt.databaseName, arr[0].databaseName)
+			assert.Equal(t, tt.tableName, arr[0].tableName)
+			assert.Equal(t, tt.isClusterTable, arr[0].isClusterTable)
+			assert.Equal(t, clusterTableSelect, arr[0].clusterTableOperation)
+		})
+	}
+}
+
 func Test_determineDML(t *testing.T) {
 	type arg struct {
 		stmt tree.Statement

pkg/frontend/databranchutils/branch_change_handle_test.go

@@ -94,6 +94,7 @@ func TestBranchChangeHandleNextReappliesPolicies(t *testing.T) {
 	require.Equal(t, engine.SnapshotReadPolicyVisibleState, engine.SnapshotReadPolicyFromContext(fake.lastCtx))
 	require.True(t, engine.RetainRowIDFromContext(fake.lastCtx))
 	require.Same(t, pkFilter, engine.PKFilterFromContext(fake.lastCtx))
+	require.False(t, engine.ChangeRangeLimitFromContext(fake.lastCtx).Enabled())
 }
 
 func TestBranchChangeHandleNextUnderlyingError(t *testing.T) {
@@ -133,6 +134,7 @@ func TestCollectChangesRange(t *testing.T) {
 			_ *mpool.MPool,
 		) (engine.ChangesHandle, error) {
 			require.Equal(t, engine.SnapshotReadPolicyVisibleState, engine.SnapshotReadPolicyFromContext(ctx))
+			require.False(t, engine.ChangeRangeLimitFromContext(ctx).Enabled())
 			return fake, nil
 		},
 	)
@@ -179,6 +181,7 @@ func TestCollectChangesPropagatesError(t *testing.T) {
 			_ *mpool.MPool,
 		) (engine.ChangesHandle, error) {
 			require.Equal(t, engine.SnapshotReadPolicyVisibleState, engine.SnapshotReadPolicyFromContext(ctx))
+			require.False(t, engine.ChangeRangeLimitFromContext(ctx).Enabled())
 			return nil, expectedErr
 		},
 	)
@@ -212,6 +215,7 @@ func TestCollectChangesWithPKFilterPropagatesPoliciesToHandle(t *testing.T) {
 			require.Equal(t, engine.SnapshotReadPolicyVisibleState, engine.SnapshotReadPolicyFromContext(ctx))
 			require.True(t, engine.RetainRowIDFromContext(ctx))
 			require.Same(t, pkFilter, engine.PKFilterFromContext(ctx))
+			require.False(t, engine.ChangeRangeLimitFromContext(ctx).Enabled())
 			return fake, nil
 		},
 	)