[CHORE]: Remove next_run from attached_functions

Author: tanujnay112

go/pkg/sysdb/coordinator/create_task_test.go

@@ -333,7 +333,6 @@ func (suite *AttachFunctionTestSuite) TestAttachFunction_IdempotentRequest_Alrea
 		OutputCollectionName:    outputCollectionName,
 		FunctionID:              functionID,
 		MinRecordsForInvocation: int64(MinRecordsForInvocation),
-		NextRun:                 now,
 		CreatedAt:               now,
 		UpdatedAt:               now,
 	}
@@ -480,7 +479,6 @@ func (suite *AttachFunctionTestSuite) TestAttachFunction_RecoveryFlow_HeapFailur
 		OutputCollectionName:    outputCollectionName,
 		FunctionID:              functionID,
 		MinRecordsForInvocation: int64(MinRecordsForInvocation),
-		NextRun:                 now,
 		CreatedAt:               now,
 		UpdatedAt:               now,
 	}
@@ -585,7 +583,6 @@ func (suite *AttachFunctionTestSuite) TestAttachFunction_IdempotentRequest_Param
 		OutputCollectionName:    outputCollectionName,
 		FunctionID:              existingOperatorID,
 		MinRecordsForInvocation: int64(MinRecordsForInvocation),
-		NextRun:                 now,
 		CreatedAt:               now,
 		UpdatedAt:               now,
 	}
@@ -665,7 +662,6 @@ func TestGetSoftDeletedAttachedFunctions_TimestampConsistency(t *testing.T) {
 			MinRecordsForInvocation: 10,
 			CreatedAt:               testTime,
 			UpdatedAt:               testTime,
-			NextRun:                 testTime,
 		},
 	}
 
@@ -702,9 +698,6 @@ func TestGetSoftDeletedAttachedFunctions_TimestampConsistency(t *testing.T) {
 	if af.UpdatedAt != expectedMicros {
 		t.Errorf("UpdatedAt timestamp mismatch: expected %d microseconds, got %d", expectedMicros, af.UpdatedAt)
 	}
-	if af.NextRunAt != expectedMicros {
-		t.Errorf("NextRunAt timestamp mismatch: expected %d microseconds, got %d", expectedMicros, af.NextRunAt)
-	}
 
 	// Verify these are NOT in seconds (would be ~1000x smaller)
 	expectedSeconds := uint64(testTime.Unix())
@@ -714,9 +707,6 @@ func TestGetSoftDeletedAttachedFunctions_TimestampConsistency(t *testing.T) {
 	if af.UpdatedAt == expectedSeconds {
 		t.Error("UpdatedAt appears to be in seconds instead of microseconds")
 	}
-	if af.NextRunAt == expectedSeconds {
-		t.Error("NextRunAt appears to be in seconds instead of microseconds")
-	}
 
 	mockMetaDomain.AssertExpectations(t)
 	mockAttachedFunctionDb.AssertExpectations(t)

go/pkg/sysdb/coordinator/list_attached_functions_test.go

@@ -64,7 +64,6 @@ func (suite *ListAttachedFunctionsTestSuite) TestListAttachedFunctions_Success()
 			DatabaseID:              "db",
 			CompletionOffset:        10,
 			MinRecordsForInvocation: 5,
-			NextRun:                 now,
 			CreatedAt:               now,
 			UpdatedAt:               now,
 		},
@@ -79,7 +78,6 @@ func (suite *ListAttachedFunctionsTestSuite) TestListAttachedFunctions_Success()
 			DatabaseID:              "db",
 			CompletionOffset:        20,
 			MinRecordsForInvocation: 15,
-			NextRun:                 now,
 			CreatedAt:               now,
 			UpdatedAt:               now,
 		},
@@ -157,7 +155,6 @@ func (suite *ListAttachedFunctionsTestSuite) TestListAttachedFunctions_FunctionD
 		DatabaseID:              "db",
 		CompletionOffset:        0,
 		MinRecordsForInvocation: 1,
-		NextRun:                 now,
 		CreatedAt:               now,
 		UpdatedAt:               now,
 	}
@@ -192,7 +189,6 @@ func (suite *ListAttachedFunctionsTestSuite) TestListAttachedFunctions_InvalidPa
 		DatabaseID:              "db",
 		CompletionOffset:        0,
 		MinRecordsForInvocation: 1,
-		NextRun:                 now,
 		CreatedAt:               now,
 		UpdatedAt:               now,
 	}

go/pkg/sysdb/coordinator/model/collection.go

@@ -99,7 +99,6 @@ type FlushCollectionInfo struct {
 	CollectionVersion        int32
 	TenantLastCompactionTime int64
 	// Optional attached function fields (only populated for attached-function-based compactions)
-	AttachedFunctionNextRun          *time.Time
 	AttachedFunctionCompletionOffset *int64
 }
 

go/pkg/sysdb/coordinator/task.go

@@ -111,7 +111,6 @@ func (s *Coordinator) AttachFunction(ctx context.Context, req *coordinatorpb.Att
 
 			// Validation passed, reuse the concurrent attached function's data
 			attachedFunctionID = concurrentAttachedFunction.ID
-			nextRun = concurrentAttachedFunction.NextRun
 			// Already created, skip Phase 2
 			skipPhase2 = true
 			return nil
@@ -187,16 +186,13 @@ func (s *Coordinator) AttachFunction(ctx context.Context, req *coordinatorpb.Att
 			FunctionParams:          paramsJSON,
 			CompletionOffset:        0,
 			LastRun:                 nil,
-			NextRun:                 now,
 			MinRecordsForInvocation: int64(req.MinRecordsForInvocation),
 			CurrentAttempts:         0,
 			CreatedAt:               now,
 			UpdatedAt:               now,
 			OldestWrittenNonce:      nil,
 		}
 
-		nextRun = attachedFunction.NextRun
-
 		err = s.catalog.metaDomain.AttachedFunctionDb(txCtx).Insert(attachedFunction)
 		if err != nil {
 			log.Error("AttachFunction: failed to insert attached function", zap.Error(err))
@@ -289,7 +285,6 @@ func attachedFunctionToProto(attachedFunction *dbmodel.AttachedFunction, functio
 		MinRecordsForInvocation: uint64(attachedFunction.MinRecordsForInvocation),
 		TenantId:                attachedFunction.TenantID,
 		DatabaseId:              attachedFunction.DatabaseID,
-		NextRunAt:               uint64(attachedFunction.NextRun.UnixMicro()),
 		CreatedAt:               uint64(attachedFunction.CreatedAt.UnixMicro()),
 		UpdatedAt:               uint64(attachedFunction.UpdatedAt.UnixMicro()),
 	}
@@ -729,7 +724,6 @@ func (s *Coordinator) GetSoftDeletedAttachedFunctions(ctx context.Context, req *
 			UpdatedAt:               uint64(af.UpdatedAt.UnixMicro()),
 		}
 
-		protoAttachedFunctions[i].NextRunAt = uint64(af.NextRun.UnixMicro())
 		if af.OutputCollectionID != nil {
 			protoAttachedFunctions[i].OutputCollectionId = proto.String(*af.OutputCollectionID)
 		}

go/pkg/sysdb/metastore/db/dbmodel/task.go

@@ -18,7 +18,6 @@ type AttachedFunction struct {
 	FunctionParams          string     `gorm:"column:function_params;type:jsonb;not null"`
 	CompletionOffset        int64      `gorm:"column:completion_offset;type:bigint;not null;default:0"`
 	LastRun                 *time.Time `gorm:"column:last_run;type:timestamp"`
-	NextRun                 time.Time  `gorm:"column:next_run;type:timestamp;not null"`
 	MinRecordsForInvocation int64      `gorm:"column:min_records_for_invocation;type:bigint;not null;default:100"`
 	CurrentAttempts         int32      `gorm:"column:current_attempts;type:integer;not null;default:0"`
 	IsAlive                 bool       `gorm:"column:is_alive;type:boolean;not null;default:true"`

go/pkg/sysdb/metastore/db/migrations/20251114134400.sql

@@ -0,0 +1,2 @@
+-- Remove next_run column from attached_functions table as it's no longer needed
+ALTER TABLE "public"."attached_functions" DROP COLUMN "next_run";

idl/chromadb/proto/coordinator.proto

@@ -575,7 +575,6 @@ message AttachedFunction {
   uint64 min_records_for_invocation = 9;
   string tenant_id = 10;
   string database_id = 11;
-  uint64 next_run_at = 12;
   uint64 created_at = 15;
   uint64 updated_at = 16;
   string function_id = 17;

rust/sysdb/src/sysdb.rs

@@ -1743,10 +1743,6 @@ impl GrpcSysDb {
             })?,
         );
 
-        // Parse next_run timestamp from microseconds
-        let next_run = std::time::SystemTime::UNIX_EPOCH
-            + std::time::Duration::from_micros(attached_function.next_run_at);
-
         // Convert params from Struct to JSON string
         let params_str = attached_function.params.map(|s| {
             let json_value = prost_struct_to_json(s);
@@ -1795,7 +1791,6 @@ impl GrpcSysDb {
             tenant_id: attached_function.tenant_id,
             database_id: attached_function.database_id,
             last_run: None,
-            next_run,
             completion_offset: attached_function.completion_offset,
             min_records_for_invocation: attached_function.min_records_for_invocation,
             is_deleted: false,

rust/sysdb/src/test_sysdb.rs

@@ -702,7 +702,6 @@ fn attached_function_to_proto(
         id: attached_function.id.0.to_string(),
         name: attached_function.name.clone(),
         function_name: attached_function.function_id.to_string(),
-        function_id: attached_function.function_id.to_string(),
         input_collection_id: attached_function.input_collection_id.0.to_string(),
         output_collection_name: attached_function.output_collection_name.clone(),
         output_collection_id: attached_function
@@ -714,9 +713,9 @@ fn attached_function_to_proto(
         min_records_for_invocation: attached_function.min_records_for_invocation,
         tenant_id: attached_function.tenant_id.clone(),
         database_id: attached_function.database_id.clone(),
-        next_run_at: system_time_to_micros(attached_function.next_run),
         created_at: system_time_to_micros(attached_function.created_at),
         updated_at: system_time_to_micros(attached_function.updated_at),
+        function_id: attached_function.function_id.to_string(),
     }
 }
 

rust/types/src/flush.rs

@@ -69,7 +69,6 @@ impl ChromaError for AdvanceAttachedFunctionError {
 
 #[derive(Debug, Clone)]
 pub struct AdvanceAttachedFunctionResponse {
-    pub next_run: std::time::SystemTime,
     pub completion_offset: u64,
 }
 
@@ -126,8 +125,6 @@ pub enum FlushCompactionResponseConversionError {
     InvalidUuid,
     #[error("Invalid attached function nonce, valid UUID required")]
     InvalidAttachedFunctionNonce,
-    #[error("Missing next_run timestamp")]
-    MissingNextRun,
     #[error("Invalid timestamp format")]
     InvalidTimestamp,
 }
@@ -139,7 +136,6 @@ impl ChromaError for FlushCompactionResponseConversionError {
             FlushCompactionResponseConversionError::InvalidAttachedFunctionNonce => {
                 ErrorCodes::InvalidArgument
             }
-            FlushCompactionResponseConversionError::MissingNextRun => ErrorCodes::InvalidArgument,
             FlushCompactionResponseConversionError::InvalidTimestamp => ErrorCodes::InvalidArgument,
             FlushCompactionResponseConversionError::DecodeError(e) => e.code(),
         }