Rebuild Python UDF with typed revisions and Arrow Flight execution

.gitignore

@@ -77,8 +77,6 @@ node-*-data/
 pkg/catalog/test/
 pkg/config/system_vars.go
 pkg/config/system_vars_test.go
-pkg/udf/pythonservice/pyserver/udf/
-pkg/udf/pythonservice/pyserver/__pycache__/
 path_to_file
 tester-log/
 ut-report/

.golangci.yml

@@ -35,20 +35,27 @@ linters:
             - $all
             - '!$test'
             - '!pkg/iceberg/adapter/iceberggo/**/*.go'
-            # Arrow IPC is isolated to the ingestion bridge and its external
-            # scan reader. Keep the dependency out of all other production
-            # packages, including the Iceberg-facing layers.
+          deny:
+            - pkg: github.com/apache/iceberg-go
+              desc: Iceberg dependency must stay behind pkg/iceberg/adapter/iceberggo
+        arrow-boundary:
+          list-mode: lax
+          files:
+            - $all
+            - '!$test'
+            - '!pkg/iceberg/adapter/iceberggo/**/*.go'
+            # Arrow belongs to the ingestion and external UDF boundaries.
+            # These owners do not also receive permission to import Iceberg.
             - '!**/pkg/container/arrowbridge/*.go'
             - '!**/pkg/sql/colexec/external/arrowio/*.go'
             - '!**/pkg/sql/colexec/external/reader_arrow.go'
             - '!**/pkg/sql/compile/compile.go'
+            - '!**/pkg/udf/python/*.go'
           deny:
-            - pkg: github.com/apache/iceberg-go
-              desc: Iceberg dependency must stay behind pkg/iceberg/adapter/iceberggo
             - pkg: github.com/apache/arrow-go
-              desc: Arrow dependency must not leak outside the Iceberg adapter
+              desc: Arrow dependency must stay behind an approved Arrow owner
             - pkg: github.com/apache/arrow/go
-              desc: Arrow dependency must not leak outside the Iceberg adapter
+              desc: Arrow dependency must stay behind an approved Arrow owner
     govet:
       disable:
         - fieldalignment

cmd/mo-service/config.go

@@ -45,7 +45,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/perfcounter"
 	"github.com/matrixorigin/matrixone/pkg/proxy"
 	"github.com/matrixorigin/matrixone/pkg/tnservice"
-	"github.com/matrixorigin/matrixone/pkg/udf/pythonservice"
+	"github.com/matrixorigin/matrixone/pkg/udf/python"
 	"github.com/matrixorigin/matrixone/pkg/util/debug/goroutine"
 	tomlutil "github.com/matrixorigin/matrixone/pkg/util/toml"
 	"github.com/matrixorigin/matrixone/pkg/version"
@@ -118,8 +118,8 @@ type Config struct {
 	CN cnservice.Config `toml:"cn"`
 	// ProxyConfig is the config of proxy.
 	ProxyConfig proxy.Config `toml:"proxy"`
-	// PythonUdfServerConfig is the config of python udf server
-	PythonUdfServerConfig pythonservice.Config `toml:"python-udf-server"`
+	// PythonUdfWorkerConfig is the config of the Python UDF worker.
+	PythonUdfWorkerConfig python.Config `toml:"python-udf-worker"`
 	// Observability parameters for the metric/trace
 	Observability config.ObservabilityParameters `toml:"observability"`
 
@@ -594,7 +594,7 @@ func (c *Config) mustGetServiceUUID() string {
 	case metadata.ServiceType_PROXY:
 		return c.ProxyConfig.UUID
 	case metadata.ServiceType_PYTHON_UDF:
-		return c.PythonUdfServerConfig.UUID
+		return c.PythonUdfWorkerConfig.UUID
 	}
 	panic("impossible")
 }

cmd/mo-service/main.go

@@ -52,7 +52,7 @@ import (
 	qclient "github.com/matrixorigin/matrixone/pkg/queryservice/client"
 	"github.com/matrixorigin/matrixone/pkg/sql/compile"
 	"github.com/matrixorigin/matrixone/pkg/tnservice"
-	"github.com/matrixorigin/matrixone/pkg/udf/pythonservice"
+	"github.com/matrixorigin/matrixone/pkg/udf/python"
 	"github.com/matrixorigin/matrixone/pkg/util/debug/goroutine"
 	"github.com/matrixorigin/matrixone/pkg/util/export"
 	"github.com/matrixorigin/matrixone/pkg/util/export/table"
@@ -294,7 +294,7 @@ func startService(
 	case metadata.ServiceType_LOG:
 		return startLogService(cfg, stopper, fs, shutdownC)
 	case metadata.ServiceType_PYTHON_UDF:
-		return startPythonUdfService(cfg, stopper)
+		return startPythonUdfWorker(cfg, stopper)
 	default:
 		panic("unknown service type")
 	}
@@ -741,8 +741,8 @@ func waitProxyFileServiceRetry(ctx context.Context, delay time.Duration) error {
 	}
 }
 
-// startPythonUdfService starts the python udf service.
-func startPythonUdfService(cfg *Config, stopper *stopper.Stopper) error {
+// startPythonUdfWorker starts the Python UDF worker.
+func startPythonUdfWorker(cfg *Config, stopper *stopper.Stopper) error {
 	if err := waitClusterCondition(cfg.mustGetServiceUUID(), cfg.HAKeeperClient, waitHAKeeperRunning); err != nil {
 		return err
 	}
@@ -755,22 +755,39 @@ func startPythonUdfService(cfg *Config, stopper *stopper.Stopper) error {
 			finish(err)
 		})
 	}
-	err := stopper.RunNamedTask("python-udf-service", func(ctx context.Context) {
+	err := stopper.RunNamedTask("python-udf-worker", func(ctx context.Context) {
 		var closeErr error
 		defer func() { finishTask(closeErr) }()
 		roleCtx, cancelRole := serviceLifecycle.roleContext(ctx, serviceRolePython)
 		defer cancelRole()
-		s, err := pythonservice.NewService(cfg.PythonUdfServerConfig)
+		s, err := python.NewSupervisor(cfg.PythonUdfWorkerConfig)
 		if err != nil {
 			panic(err)
 		}
 		if err := s.Start(); err != nil {
 			panic(err)
 		}
-		<-roleCtx.Done()
-		if err := s.Close(); err != nil {
-			closeErr = err
-			logutil.GetGlobalLogger().Error("failed to close python udf service", zap.Error(err))
+		workerDone := s.Done()
+		select {
+		case <-roleCtx.Done():
+			if err := s.Close(); err != nil {
+				closeErr = err
+				logutil.GetGlobalLogger().Error("failed to close Python UDF worker", zap.Error(err))
+			}
+		case <-workerDone:
+			// A worker exit while the role is still running leaves CN with a
+			// configured but unusable Python runtime. Surface it as a fatal
+			// service event so the existing lifecycle supervisor drains CN
+			// before the worker dependency and does not accept partial service
+			// availability. A zero exit is still unexpected here.
+			if roleCtx.Err() == nil {
+				workerErr := s.Err()
+				if workerErr == nil {
+					workerErr = errors.New("process exited without an error")
+				}
+				closeErr = fmt.Errorf("python UDF worker exited unexpectedly: %w", workerErr)
+				serviceLifecycle.notifyFatal(closeErr)
+			}
 		}
 	})
 	if err != nil {

docs/cn/stream_transport_fast_upload_plan.md

@@ -20,7 +20,7 @@ Phase 0 — Protobuf / gRPC contract
 
 New file `proto/datastream/v1/datastream.proto` (proto3, **not** gogoproto —
 this proto is shared with Java, so use standard `protoc-gen-go` +
-`protoc-gen-go-grpc`, same toolchain as `pkg/udf/udf.proto`, which is the
+`protoc-gen-go-grpc`, using the repository's current protobuf toolchain, which is the
 existing real-gRPC precedent in the repo; the main `proto/*.proto` files are
 gogo-generated and Java-hostile).
 

docs/design/python_udf.md

@@ -0,0 +1,112 @@
+# Python UDF implementation and rollout scope
+
+This describes the implemented, owner-authorized external Python adapter and its
+PR acceptance boundary. The relevant current-stage Catalog, SQL/planner,
+migration/restore, Python ABI, Flight, and resource contracts are consolidated
+in the [versioned current-stage design](python_udf_current_stage.md). Broader
+long-term designs continue to inform future work; this page does not approve a
+production tenant-isolation model.
+
+## Owner-approved stage scope
+
+On 2026-09-24, feature owner `iamlinjunhong` approved this PR for testing and
+development-stage acceptance. The user guide and SQL examples are recorded in
+[issue #28132](https://github.com/matrixorigin/matrixone/issues/28132#issuecomment-5748253137).
+This PR does not claim production rollout, sandbox isolation, protection for
+untrusted Python, or authenticated/TLS Flight transport. Those remain separate
+production acceptance gates. Python stays disabled in generic launch and requires
+explicit opt-in in the test/development configurations.
+
+This stage decision is the feature owner's authorization for the current PR
+scope; it is not a claim that Architecture, Security, SQL/Planner, Runtime,
+Catalog, or Operator owners have approved the future production design.
+
+## Execution and identity
+
+CREATE/REPLACE validates Python syntax and the handler contract before publishing
+a definition. The shared Catalog stores immutable typed revisions and complete
+argument/return descriptors. Artifacts are digest-addressed, account-scoped and
+published through FileService; worker execution receives the verified artifact
+from CN. The worker needs no object-storage credentials or query-time package
+installation. Old demo definitions are inert: explicitly recreate them using the
+current DDL and SDK. Unknown definition, plan, Arrow, SDK or wire contracts fail
+before user execution; no legacy fallback or automatic adapter is provided.
+
+Planning resolves an exact FunctionRef and RoutineCall. Prepared/cache reuse
+validates the selected revision and the complete overload namespace state; see
+[routine namespace validation](routine_namespace_validation.md). Execution never
+resolves a mutable latest revision. Clone and restore preserve supported catalog
+contracts and republish Python artifacts for the destination account.
+
+CN keeps native MO vectors. The expression framework dispatches typed Python calls
+to ExternalRoutineEval; the parent evaluator owns argument evaluation, casts,
+CASE/selection and NULL guards. The physical stage compacts selected rows,
+admits bounded asynchronous work and scatters verified results, preserving empty
+input, NULLs, zero arguments, nested calls and default VOLATILE semantics.
+
+The Gateway is a CN library, not a separately deployed service. It connects to
+Arrow Flight served by the Python worker. No second Flight daemon is required.
+A worker creates bounded handler subprocesses; each handler/module can be reused
+only within the invocation's bounded burst. It does not pool user heaps across
+invocations. SCALAR handlers receive `(ctx, scalar...)`; VECTOR handlers receive
+`(ctx, arrays...)` and use `ctx.num_rows` for output shape. Arrow field/schema
+metadata comes from the trusted adapter, not handler-supplied identity.
+
+## Completion and resources
+
+K bounds active Gateway invocations, H bounds handler execution slots, and W=1
+bounds stream batch progress. Admission, group/member state, terminal records and
+per-account/per-owner quotas have independent owners and budgets. Protocol steps
+include EndInput, InputConsumed, result ACK and Finish/FinishAck. Started work is
+never transparently replayed. Cancellation and capability refresh waiters progress
+independently; generation fences reject stale completions.
+
+Process cleanup must cover both termination and reaping. Handler sessions own
+pipes, child/watchdog processes and H/quota until cleanup succeeds; partial
+initialization transfers a failed session to the same bounded pending-cleanup
+owner. Linux containers run `/usr/bin/tini -- python -u worker.py`; init adopts
+orphaned descendants after their handler leader exits. An Operator command override
+must retain this init process. Native Supervisor launches rely on the host's init
+(or an explicit subreaper in a container test harness).
+
+## Enablement and deployment boundary
+
+Generic launch keeps Python disabled. The explicit worker-enabled launch and
+ordinary BVT entry point enable the unisolated adapter. See
+[Python BVT and local launch](../../test/distributed/cases/udf_python/README.md).
+Compose configures a separate worker endpoint for each CN. The `PYTHON_UDF`
+mo-service role wraps a Go Supervisor; the standalone worker image runs Python
+under init. These are different launch forms for the same Flight service.
+
+Gateway and worker must match the current capability contract, including tzdata.
+The worker image replaces its entire zoneinfo tree from the MO runtime base;
+upgrading tzdata also changes the environment digest of definitions. Check existing
+definitions before enabling an incompatible environment and explicitly rebuild
+where required. Matching software version strings alone are not this check.
+
+`enabled` and `allow-unisolated` are explicit gates. This implementation does not
+provide sandbox isolation or authenticated/TLS Flight transport. Same-Pod Operator
+configuration binds the worker to loopback and controls Pod composition; it does
+not add broad CN ingress grants. External worker exposure needs a separately
+approved security/deployment design. Worker availability is a Python execution
+condition, not evidence of production tenant isolation.
+
+## Verification and remaining scope
+
+The regression suite covers scalar/vector Arrow computation, types and malformed
+values, guarded SQL positions and DML atomicity, revision/overload invalidation,
+exact-descriptor DROP and cross-account snapshot restore. Runtime tests cover
+ACK/Finish, cancellation, quotas, initialization/cleanup failure, worker death,
+stale events and real Flight consumers. Operator status wire fixtures pin the
+producer/consumer error contract.
+
+Linux PR validation uses the repository CGo wrapper, two CNs with distinct
+Supervisor/worker endpoints, ordinary SQL/Python BVT, current worker image and
+process-tree fault injection. Record the exact revision, selected tests, image
+and terminal result with each run; unit or CNI echo tests are not complete
+Operator rollout evidence.
+
+Sandbox, production deployment approval, imported dependency environments and
+W>1/cumulative ACK enablement remain separate work. W=1 measurements do not prove
+that a wider window has no benefit. This PR is related to issue #28132 and does
+not close that issue's broader isolated-runtime acceptance criteria.