docs/test: document Spark JDBC READ_COMMITTED and TopN ANY NULL

README.md

@@ -248,6 +248,7 @@ Ready to dive deeper? Explore our comprehensive collection of hands-on tutorials
 | Tutorial | Language/Framework | Description |
 |----------|-------------------|-------------|
 | [Java CRUD Demo](https://docs.matrixorigin.cn/mo/en/latest/MatrixOne/Tutorial/develop-java-crud-demo.html) | Java | Java application development |
+| [Spark JDBC writes](docs/guides/spark-jdbc.md) | Python / Spark | Configure Spark JDBC writes for MatrixOne's supported isolation level |
 | [SpringBoot and JPA CRUD Demo](https://docs.matrixorigin.cn/mo/en/latest/MatrixOne/Tutorial/springboot-hibernate-crud-demo.html) | Java | SpringBoot with Hibernate/JPA |
 | [PyMySQL CRUD Demo](https://docs.matrixorigin.cn/mo/en/latest/MatrixOne/Tutorial/develop-python-crud-demo.html) | Python | Basic database operations with Python |
 | [SQLAlchemy CRUD Demo](https://docs.matrixorigin.cn/mo/en/latest/MatrixOne/Tutorial/sqlalchemy-python-crud-demo.html) | Python | Python with SQLAlchemy ORM |

README_CN.md

@@ -242,6 +242,7 @@ for row in results.rows:
 | 教程 | 语言/框架 | 说明 |
 |----------|-------------------|-------------|
 | [Java CRUD 示例](https://docs.matrixorigin.cn/mo/en/latest/MatrixOne/Tutorial/develop-java-crud-demo.html) | Java | Java 应用开发 |
+| [Spark JDBC 写入指南](docs/guides/spark-jdbc.md) | Python / Spark | 为 MatrixOne 配置受支持的 JDBC 写入隔离级别 |
 | [SpringBoot 和 JPA CRUD 示例](https://docs.matrixorigin.cn/mo/en/latest/MatrixOne/Tutorial/springboot-hibernate-crud-demo.html) | Java | SpringBoot + Hibernate/JPA |
 | [PyMySQL CRUD 示例](https://docs.matrixorigin.cn/mo/en/latest/MatrixOne/Tutorial/develop-python-crud-demo.html) | Python | Python 基础数据库操作 |
 | [SQLAlchemy CRUD 示例](https://docs.matrixorigin.cn/mo/en/latest/MatrixOne/Tutorial/sqlalchemy-python-crud-demo.html) | Python | Python + SQLAlchemy ORM |

docs/guides/spark-jdbc.md

@@ -0,0 +1,53 @@
+# Spark JDBC writes to MatrixOne
+
+[Spark's JDBC write option documentation](https://spark.apache.org/docs/3.5.6/sql-data-sources-jdbc.html#data-source-option)
+lists `READ_UNCOMMITTED` as the default `isolationLevel`. MatrixOne rejects
+that isolation level, so a Spark write that leaves the default in place fails
+when Connector/J configures the writer connection. Set Spark's JDBC
+`isolationLevel` option to `READ_COMMITTED` on every write:
+
+```python
+connection_properties = {
+    "driver": "com.mysql.cj.jdbc.Driver",
+    "user": "your_user",
+    "password": "your_password",
+    "isolationLevel": "READ_COMMITTED",
+}
+
+df.write.jdbc(
+    url="jdbc:mysql://127.0.0.1:6001/your_database",
+    table="target_table",
+    mode="append",
+    properties=connection_properties,
+)
+```
+
+`isolationLevel` is a Spark JDBC **write** option. Spark opens JDBC connections
+for its writer tasks, so setting isolation on a separate SQL connection does
+not configure those writes. MatrixOne supports `READ_COMMITTED`; its session
+isolation variable displays this as `READ-COMMITTED`.
+
+Read the committed output with a new JDBC read when checking a pipeline:
+
+```python
+read_properties = {
+    "driver": "com.mysql.cj.jdbc.Driver",
+    "user": "your_user",
+    "password": "your_password",
+}
+readback = spark.read.jdbc(
+    url="jdbc:mysql://127.0.0.1:6001/your_database",
+    table="target_table",
+    properties=read_properties,
+)
+```
+
+The repository includes a runnable compatibility probe that checks the default
+failure, the explicit `READ_COMMITTED` write across two writer partitions, and
+readback through both a fresh Spark JDBC read and a separate SQL connection. It
+places one row in each writer partition to exercise both JDBC writer tasks:
+[`test/manual/spark_jdbc/README.md`](../../test/manual/spark_jdbc/README.md).
+
+This option selects a supported isolation level for each Spark write. It does
+not make a multi-partition Spark append atomic or change Spark's task retry
+behavior.

pkg/sql/colexec/partition/topn_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/container/nulls"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
@@ -760,3 +761,65 @@ func collectTopNRowsAt(
 	}
 	return rows
 }
+
+func TestPartitionTopNNonLiteralAnyNullOrder(t *testing.T) {
+	proc := testutil.NewProcessWithOwnedMPool(t, "", mpool.MustNewZero())
+	t.Cleanup(func() { require.Zero(t, proc.Mp().CurrNB()) })
+	t.Cleanup(proc.Free)
+	arg := newTopNArgument(1)
+	arg.OrderBySpecs[1].Expr = topNCol(1, types.T_any)
+	t.Cleanup(func() { arg.Free(proc, false, nil) })
+
+	for generation := 0; generation < 2; generation++ {
+		batches := []*batch.Batch{
+			makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{10, 20}, false),
+			makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{11, 21}, true),
+		}
+		child := colexec.NewMockOperator().WithBatchs(batches)
+		t.Cleanup(func() { child.Free(proc, false, nil) })
+		arg.Children = nil
+		arg.AppendChild(child)
+
+		require.NoError(t, arg.Prepare(proc))
+		require.Len(t, arg.top.compares, 2)
+		require.NotNil(t, arg.top.compares[0])
+
+		counts := make(map[int32]int)
+		validPayload := map[int32][]int64{1: {10, 11}, 2: {20, 21}}
+		for {
+			result, err := arg.Call(proc)
+			require.NoError(t, err)
+			if result.Batch == nil || result.Status == vm.ExecStop {
+				break
+			}
+			groups := vector.MustFixedColWithTypeCheck[int32](result.Batch.Vecs[0])
+			payloads := vector.MustFixedColWithTypeCheck[int64](result.Batch.Vecs[2])
+			for row, group := range groups {
+				require.True(t, result.Batch.Vecs[1].IsNull(uint64(row)))
+				require.Contains(t, validPayload[group], payloads[row])
+				counts[group]++
+			}
+		}
+		require.Equal(t, map[int32]int{1: 1, 2: 1}, counts)
+		arg.Reset(proc, false, nil)
+	}
+}
+
+func makeAnyNullTopNBatch(t *testing.T, proc *process.Process, groups []int32, payloads []int64, constant bool) *batch.Batch {
+	t.Helper()
+	require.Len(t, groups, len(payloads))
+	bat := batch.NewWithSize(3)
+	bat.Vecs[0] = vector.NewVec(types.T_int32.ToType())
+	require.NoError(t, vector.AppendFixedList(bat.Vecs[0], groups, nil, proc.Mp()))
+	if constant {
+		bat.Vecs[1] = vector.NewConstNull(types.T_any.ToType(), len(groups), proc.Mp())
+	} else {
+		bat.Vecs[1] = vector.NewVec(types.T_any.ToType())
+		bat.Vecs[1].SetLength(len(groups))
+		nulls.AddRange(bat.Vecs[1].GetNulls(), 0, uint64(len(groups)))
+	}
+	bat.Vecs[2] = vector.NewVec(types.T_int64.ToType())
+	require.NoError(t, vector.AppendFixedList(bat.Vecs[2], payloads, nil, proc.Mp()))
+	bat.SetRowCount(len(groups))
+	return bat
+}

test/manual/spark_jdbc/README.md

@@ -0,0 +1,46 @@
+# Spark JDBC isolation probe
+
+This probe reproduces the Spark JDBC write-isolation compatibility boundary
+with Spark 3.5.3 and MySQL Connector/J 8.0.33. It verifies that a writer with
+Spark's default isolation fails with MatrixOne's explicit unsupported-level
+error and inserts no rows, then writes the same two rows with
+`isolationLevel=READ_COMMITTED` across two non-empty writer partitions and reads
+them back through Spark and a separate JDBC connection. The probe creates uniquely named
+tables and removes them when it exits.
+
+Use an isolated MatrixOne instance and a dedicated database. Set the connection
+environment variables without placing credentials in the source:
+
+```sh
+export MO_JDBC_URL='jdbc:mysql://127.0.0.1:6001/spark_jdbc_issue_29838'
+export MO_USER='your_user'
+export MO_PASSWORD='your_password'
+export MYSQL_CONNECTOR_JAR='/absolute/path/to/mysql-connector-j-8.0.33.jar'
+```
+
+Create the database before running the probe. Install the pinned PySpark and
+Connector/J dependencies, then launch it with both driver and executor
+classpath entries for Connector/J:
+
+```sh
+python3 -m venv /tmp/spark-jdbc-29838-venv
+/tmp/spark-jdbc-29838-venv/bin/pip install pyspark==3.5.3
+curl -fL \
+  https://repo.maven.apache.org/maven2/com/mysql/mysql-connector-j/8.0.33/mysql-connector-j-8.0.33.jar \
+  -o /tmp/mysql-connector-j-8.0.33.jar
+export MYSQL_CONNECTOR_JAR=/tmp/mysql-connector-j-8.0.33.jar
+export PYSPARK_DRIVER_PYTHON=/tmp/spark-jdbc-29838-venv/bin/python
+export PYSPARK_PYTHON=/tmp/spark-jdbc-29838-venv/bin/python
+
+/tmp/spark-jdbc-29838-venv/bin/spark-submit \
+  --master 'local[2]' \
+  --jars "$MYSQL_CONNECTOR_JAR" \
+  --driver-class-path "$MYSQL_CONNECTOR_JAR" \
+  test/manual/spark_jdbc/spark_jdbc_isolation_probe.py
+```
+
+The command can be run again to check reuse and cleanup. It prints the MatrixOne
+version, Spark version, and the verified default failure and `READ_COMMITTED`
+readback. The probe fails if the default write succeeds, if an unrelated error
+causes the default write to fail, if any default rows were inserted, or if the
+explicit-isolation write/readback differs from the expected rows.