Apache Hudi through Spark Connect: running the full table surface from a thin client

• Architecture: what Spark Connect changes for Hudi
• Starting a server that speaks Hudi
  • Two server settings the bundle does not bring with it
  • The extension class name, which is the usual mistake
• Connecting the client
• Copy-on-write tables, the full DML surface
  • The partitioning trap
  • Update, merge and delete
• Schema evolution: one of four on defaults, four of four with two configs
• Merge-on-read, and what the two query types show
• Incremental queries and time travel
• Procedures and metadata, over gRPC
  • rollback_to_instant only unwinds the newest instant, and not always that
• What the thin client cannot do
• Common misconceptions
• Production notes
• Frequently asked questions
• Conclusion
• References
• Trademarks

TL;DR

• Spark Connect moves planning to a server, so the Hudi bundle, spark.sql.extensions, the Kryo serializer and HoodieCatalog all belong to the server at startup. A client cannot add them later.
• The client becomes a pip install with no Spark, no Java and no Hudi jar. Everything below was driven from one.
• Almost the whole surface works: COW and MOR, partitioned tables, INSERT, UPDATE, MERGE, DELETE, incremental queries, time travel, 12 stored procedures and the metadata table functions.
• Four things do not, at least not on defaults. A TIMESTAMP precombine field breaks UPDATE until the server sets one Parquet-Avro option, RENAME COLUMN, DROP COLUMN and type widening are rejected until two schema-evolution configs are set, rollback_to_instant only unwinds the newest instant and refuses even that once the metadata table has compacted past it, and hudi_table_changes in CDC mode needs a CDC-enabled table.
• The trap that costs the most time is not a Connect issue at all: partitioning moves the partition column to the end of the schema, so a positional INSERT fails with an error that blames the wrong column.

Hudi is not part of Spark. It arrives as a bundle jar, installs a Catalyst extension, registers a catalog and needs a specific serializer. All four are session-level concerns, and Spark Connect moves the session to the far side of a gRPC connection.

So the practical question is not whether Hudi works through Spark Connect. It is which half of your setup goes where, and what the failures look like when a piece lands on the wrong side. This post answers that by running the whole thing: one Connect server, one client with nothing installed, and every Hudi operation worth using, including the ones that fail.

Every result quoted here came out of that run. Where something failed I say so, and where a failure turned out to be my own mistake rather than a limitation I say that too, because the error messages in two of those cases point somewhere misleading.

Architecture: what Spark Connect changes for Hudi

In a classic session your process owns the SparkSession, the analyzer, the catalog and the scheduler. Hudi’s extension and catalog are installed into your session, by your own spark-submit line.

Spark Connect cuts that in half:

flowchart TB
  subgraph C["client, your laptop"]
    A["DataFrame and SQL calls"] --> B["unresolved plan<br/>as protobuf"]
  end
  B -->|"gRPC, port 15002"| S
  subgraph S["Spark Connect server, the driver"]
    D["analyzer plus HoodieCatalog"] --> E["Catalyst optimiser plus<br/>HoodieSparkSessionExtension"]
    E --> F["physical plan"]
  end
  F --> X["executors, Hudi bundle on the classpath"]
  X -->|"Arrow batches"| A

The client cannot resolve default.trips_cow because it has no catalog and no idea what a Hudi table is. It sends the operation and lets the server work it out. Everything below follows from that single split.

Hudi plugs into Spark at four points, and Connect puts all four on the server:

flowchart TB
  subgraph CL["client"]
    P["pip install pyspark<br/>no Java, no Hudi jar"]
  end
  subgraph SV["server, fixed at startup"]
    J["hudi-spark4.1-bundle<br/>plus parquet-variant"]
    K["HoodieSparkSessionExtension"]
    L["HoodieCatalog over spark_catalog"]
    M["KryoSerializer"]
  end
  P -->|"SQL and DataFrame"| SV
  J --- K --- L --- M

None of the four can be set by a client. spark.conf.set reaches the server for runtime SQL configs, but spark.sql.extensions is read once while the session is built, before any client connects. A client can read it, which is the fastest way to tell a broken server from a broken query.

Starting a server that speaks Hudi

Use --packages when the host can reach Maven Central, --jars with pre-downloaded files when it cannot. The run behind this post used Spark 4.1.3 and --packages:

$SPARK_HOME/sbin/start-connect-server.sh \
  --packages org.apache.hudi:hudi-spark4.1-bundle_2.13:1.2.0,org.apache.parquet:parquet-variant:1.16.0 \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.hadoop.parquet.avro.data.supplier=org.apache.parquet.avro.GenericDataSupplier \
  --conf spark.sql.warehouse.dir=/tmp/warehouse/spark \
  --master "local[4]"

In a container, set SPARK_NO_DAEMONIZE=1 so the script stays in the foreground, and add --conf spark.jars.ivy=/tmp/ivy if the default Ivy directory is not writable.

The coordinate encodes two compatibilities at once, and both are hard failures rather than warnings:

Spark 4 is Scala 2.13 only, so a _2.12 bundle never loads.

Two server settings the bundle does not bring with it

The last line of the coordinate and the parquet.avro.data.supplier conf are not decoration. With only the bundle on a Spark 4.1.3 server, CREATE TABLE and INSERT succeed, and the first UPDATE fails inside the write task:

HoodieUpsertException: Error upserting bucketType UPDATE for partition :0
Caused by: java.lang.NoClassDefFoundError: org/apache/parquet/variant/VariantConverters$ParentConverter

The bundle ships parquet-avro 1.16.0, whose AvroReadSupport loads classes from the separate parquet-variant artifact. Spark 4.1.3 ships the other Parquet 1.16.0 jars but not that one, and neither does the bundle. Adding org.apache.parquet:parquet-variant:1.16.0 to --packages clears it.

That exposes the second failure, on the same UPDATE:

HoodieUpsertException: Error upserting bucketType UPDATE for partition :0
Caused by: java.lang.ClassCastException: class java.time.Instant cannot be cast to class java.lang.Long

The stack trace ends in Hudi’s event-time merger, which compares the preCombineField of the stored row with the incoming one. The stored value is read back as a java.time.Instant, the incoming one is a Long, and the comparison throws. It only happens when the precombine field is a TIMESTAMP: the same UPDATE on a table ordered by a BIGINT succeeded on the same server. Setting spark.hadoop.parquet.avro.data.supplier to org.apache.parquet.avro.GenericDataSupplier was the one change that made the TIMESTAMP case work, and every UPDATE, MERGE and upsert below ran with it.

Moving to the 1.2.1 bundle is not a shortcut. It does contain the parquet-variant classes, but the same UPDATE failed on it with the same ClassCastException.

The extension class name, which is the usual mistake

Hudi’s session extension is org.apache.spark.sql.hudi.HoodieSparkSessionExtension. It lives under org.apache.spark.sql, not under org.apache.hudi, which is where most people reach for it first, and where I reached for it first.

Getting it wrong does not stop the server. That is what makes it expensive: the server starts, reports healthy, and then every MERGE and CALL fails later with an error about syntax or a missing table. The only evidence is a warning in the server log at startup:

WARN SparkSession: Cannot use org.apache.hudi.HoodieSparkSessionExtension to configure session extensions.
java.lang.ClassNotFoundException: org.apache.hudi.HoodieSparkSessionExtension

Two habits remove this whole class of problem. Read the name out of the jar rather than from memory:

jar tf hudi-spark4.1-bundle_2.13-1.2.0.jar | grep SessionExtension
# org/apache/spark/sql/hudi/HoodieSparkSessionExtension.class

And confirm from the client that the server loaded it:

print(spark.conf.get("spark.sql.extensions"))
# org.apache.spark.sql.hudi.HoodieSparkSessionExtension

That one line separates “my SQL is wrong” from “the server is misconfigured”, and it is the first thing to run when a MERGE is rejected.

Connecting the client

The client needs pyspark and gRPC. No Java, no SPARK_HOME, no Hudi jar. Pin it to the version your server runs:

pip install "pyspark[connect]==4.1.3"

from pyspark.sql import SparkSession

spark = SparkSession.builder.remote("sc://localhost:15002").getOrCreate()
print(spark.version)

spark.version reports the server’s version, so it tells you nothing about your client. Check the client separately with pyspark.__version__.

Match the client’s Python minor version to the server’s as well. The client here ran in a python:3.10-slim container because the apache/spark:4.1.3-python3 server image runs Python 3.10, and Python UDFs break when the two minors differ.

The pin matters more than it looks, and a bare pip install "pyspark[connect]" is the wrong instruction. It installs whatever is newest on PyPI, and a client newer than the server can fail on its first createDataFrame. A 4.1.3 client against a 3.5.9 server did exactly that: createDataFrame failed with [SQL_CONF_NOT_FOUND] for spark.sql.session.localRelationChunkSizeRows.

The newer client asks the older server for a SQL config that does not exist there, and only the paths that build a local relation from client data trip over it, so the failure looks arbitrary.

The gap is not symmetric. An older client against a newer server was tolerated: a 3.5.9 client against the 4.1.3 Hudi server created a table and ran INSERT, UPDATE, MERGE, CALL show_commits, createDataFrame and spark.range without complaint. A newer client against an older server was not. Match the versions, and when you cannot, be the older side.

Copy-on-write tables, the full DML surface

Hudi installs over the default catalog rather than adding a named one, so tables are ordinary two-part names. The Hudi specifics go in TBLPROPERTIES.

spark.sql("""
  CREATE TABLE default.trips_cow (
    trip_id    BIGINT,
    rider_id   BIGINT,
    city       STRING,
    fare       DECIMAL(10,2),
    started_at TIMESTAMP)
  USING hudi
  PARTITIONED BY (city)
  TBLPROPERTIES (
    primaryKey      = 'trip_id',
    preCombineField = 'started_at',
    type            = 'cow')
  LOCATION '/tmp/warehouse/trips_cow'""")

primaryKey is what makes an UPDATE or MERGE possible at all, since Hudi needs a record key to find the row it is replacing. preCombineField decides which version wins when two records share a key. type = 'cow' selects copy-on-write, where an update rewrites the whole base file, as opposed to merge-on-read, which appends a delta log and merges at query time.

The partitioning trap

Now the thing that will cost you an afternoon. This insert looks obviously correct and fails:

spark.sql("""
  INSERT INTO default.trips_cow VALUES
    (1, 101, 'Hyderabad', 245.50, TIMESTAMP '2026-09-01 08:15:00')""")

AnalysisException: [INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST]
Cannot write incompatible data for the table `spark_catalog`.`default`.`trips_cow`:
Cannot safely cast `fare` "STRING" to "DECIMAL(10,2)". SQLSTATE: KD000

The error names fare, and fare is fine. Look at the schema Hudi actually created:

[f.name for f in spark.table("default.trips_cow").schema.fields
 if not f.name.startswith("_hoodie")]

['trip_id', 'rider_id', 'fare', 'started_at', 'city']

Partitioning moves the partition column to the end of the schema. The table was declared with city third, and it is stored fifth. A positional INSERT ... VALUES maps by position, so 'Hyderabad' lands on fare, and the cast error is a truthful report of a problem one column to the left of where you are looking.

Two fixes. Name the columns, which is worth doing on any Hudi table whether it is partitioned or not:

spark.sql("""
  INSERT INTO default.trips_cow (trip_id, rider_id, city, fare, started_at) VALUES
    (1, 101, 'Hyderabad', 245.50, TIMESTAMP '2026-09-01 08:15:00'),
    (2, 102, 'Bengaluru', 180.00, TIMESTAMP '2026-09-01 09:30:00'),
    (3, 103, 'Hyderabad', 320.75, TIMESTAMP '2026-09-02 07:45:00')""")

Or supply the values in Hudi’s stored order, with the partition column last. Both were verified; the explicit column list is the one to use, because it survives a later ADD COLUMNS.

Update, merge and delete

spark.sql("UPDATE default.trips_cow SET fare = 250.00 WHERE trip_id = 1")

spark.sql("""
  MERGE INTO default.trips_cow t
  USING (SELECT 4L AS trip_id, 104L AS rider_id, 'Pune' AS city,
                CAST(99.00 AS DECIMAL(10,2)) AS fare,
                TIMESTAMP '2026-09-03 10:00:00' AS started_at) s
    ON t.trip_id = s.trip_id
  WHEN MATCHED     THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *""")

spark.sql("DELETE FROM default.trips_cow WHERE trip_id = 2")

All three are extension-provided syntax, so they are also a working test that the server loaded HoodieSparkSessionExtension. Results from the sequence above:

count after INSERT                      3
fare for trip_id 1 after UPDATE         250.00
count after MERGE (Pune inserted)       4
count after DELETE                      3
count WHERE city = 'Hyderabad'          2
SHOW PARTITIONS   city=Bengaluru, city=Hyderabad, city=Pune

SHOW PARTITIONS still listing city=Bengaluru after its only row was deleted is correct: the partition directory and its metadata remain until cleaning removes them.

The five Hudi bookkeeping columns come through the client untouched, which is a quick way to confirm you are reading a real Hudi table and not a directory of Parquet:

[c for c in spark.table("default.trips_cow").columns if c.startswith("_hoodie")]

['_hoodie_commit_time', '_hoodie_commit_seqno', '_hoodie_record_key',
 '_hoodie_partition_path', '_hoodie_file_name']

Schema evolution: one of four on defaults, four of four with two configs

ADD COLUMNS works, and old rows read back NULL for the new column as they should:

spark.sql("ALTER TABLE default.trips_cow ADD COLUMNS (pay_method STRING)")

spark.sql("""
  INSERT INTO default.trips_cow (trip_id, rider_id, city, fare, started_at, pay_method)
  VALUES (5, 105, 'Chennai', 75.25, TIMESTAMP '2026-09-04 11:00:00', 'upi')""")

pay_method for trip_id 5           upi
rows WHERE pay_method IS NULL      3

On default settings the other three are refused:

spark.sql("ALTER TABLE default.trips_cow RENAME COLUMN pay_method TO pm")
spark.sql("ALTER TABLE default.trips_cow DROP COLUMN pay_method")

AnalysisException: [UNSUPPORTED_FEATURE.TABLE_OPERATION] The feature is not supported:
Table `spark_catalog`.`default`.`trips_cow` does not support RENAME COLUMN. Please check
the current catalog and namespace to make sure the qualified table name is expected, and
also check the catalog implementation which is configured by "spark.sql.catalog". SQLSTATE: 0A000

AnalysisException: [UNSUPPORTED_FEATURE.TABLE_OPERATION] The feature is not supported:
Table `spark_catalog`.`default`.`trips_cow` does not support DROP COLUMN. Please check
the current catalog and namespace to make sure the qualified table name is expected, and
also check the catalog implementation which is configured by "spark.sql.catalog". SQLSTATE: 0A000

Widening a type is refused too. On a small table with an INT and a FLOAT column:

spark.sql("""
  CREATE TABLE default.widen_t (id BIGINT, qty INT, rate FLOAT, ts TIMESTAMP)
  USING hudi
  TBLPROPERTIES (primaryKey = 'id', preCombineField = 'ts', type = 'cow')
  LOCATION '/tmp/warehouse/widen_t'""")
spark.sql("""INSERT INTO default.widen_t (id, qty, rate, ts)
             VALUES (1, 5, 1.5, TIMESTAMP '2026-09-01 00:00:00')""")

spark.sql("ALTER TABLE default.widen_t ALTER COLUMN qty TYPE BIGINT")
spark.sql("ALTER TABLE default.widen_t ALTER COLUMN rate TYPE DOUBLE")

AnalysisException: ALTER TABLE CHANGE COLUMN is not supported for changing column 'qty'
with type 'IntegerType' to 'qty' with type 'LongType'

AnalysisException: ALTER TABLE CHANGE COLUMN is not supported for changing column 'rate'
with type 'FloatType' to 'rate' with type 'DoubleType'

None of this is a Spark Connect limit. Hudi gates rename, drop and type changes behind schema-on-read evolution, and both switches are runtime configs, so a client can set them for its own session:

spark.conf.set("hoodie.schema.on.read.enable", "true")
spark.conf.set("hoodie.datasource.write.schema.allow.auto.evolution.column.drop", "true")

spark.sql("ALTER TABLE default.trips_cow RENAME COLUMN pay_method TO pm")
spark.sql("ALTER TABLE default.trips_cow DROP COLUMN pm")
spark.sql("ALTER TABLE default.widen_t ALTER COLUMN qty TYPE BIGINT")
spark.sql("ALTER TABLE default.widen_t ALTER COLUMN rate TYPE DOUBLE")

spark.conf.unset("hoodie.schema.on.read.enable")
spark.conf.unset("hoodie.datasource.write.schema.allow.auto.evolution.column.drop")

columns after RENAME   ['trip_id', 'rider_id', 'fare', 'started_at', 'pm', 'city']
pm for trip_id 5       upi
columns after DROP     ['trip_id', 'rider_id', 'fare', 'started_at', 'city']
widen_t after ALTER    qty BIGINT, rate DOUBLE, row still (5, 1.5)

Schema-on-read alone is not enough: with only the first config set, the rename and the drop both failed with MissingSchemaFieldException ... Fields missing from incoming schema. The second config is what lets the rewritten schema lose a field.

The unset at the end matters on this table. While hoodie.schema.on.read.enable was still true, a plain spark.table("default.trips_cow").count() on the partitioned table failed:

java.lang.ClassCastException: class org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
cannot be cast to class org.apache.spark.sql.execution.vectorized.ConstantColumnVector

After the two unset calls the same count returned 4. Treat the configs as something to switch on around an ALTER, not a session default.

Merge-on-read, and what the two query types show

Merge-on-read is where Hudi’s read path becomes visible, and the divergence is easy to demonstrate through a thin client:

spark.sql("""
  CREATE TABLE default.trips_mor (
    trip_id BIGINT, rider_id BIGINT, city STRING,
    fare DECIMAL(10,2), started_at TIMESTAMP)
  USING hudi
  TBLPROPERTIES (primaryKey = 'trip_id', preCombineField = 'started_at', type = 'mor')
  LOCATION '/tmp/warehouse/trips_mor'""")

spark.sql("""
  INSERT INTO default.trips_mor VALUES
    (1, 101, 'Hyderabad', 245.50, TIMESTAMP '2026-09-01 08:15:00'),
    (2, 102, 'Bengaluru', 180.00, TIMESTAMP '2026-09-01 09:30:00')""")

spark.sql("UPDATE default.trips_mor SET fare = 999.99 WHERE trip_id = 1")

Now read the same row two ways:

# snapshot, the default: base files merged with the delta logs
spark.sql("SELECT fare FROM default.trips_mor WHERE trip_id = 1").collect()[0][0]

# read-optimised: base files only, ignoring the logs
(spark.read.format("hudi")
      .option("hoodie.datasource.query.type", "read_optimized")
      .load("/tmp/warehouse/trips_mor")
      .filter("trip_id = 1").select("fare").collect()[0][0])

snapshot         999.99
read_optimized   245.50

That gap is the whole merge-on-read trade, measured: the snapshot read pays a merge to be correct, the read-optimised read is cheaper and stale until compaction. Both paths work identically through Connect.

Compaction is where expectations need managing. run_compaction is not a command that compacts on demand:

spark.sql("CALL run_compaction(op => 'scheduleandexecute', table => 'default.trips_mor')")

0 rows returned, and CALL show_compaction(...) reports 0 pending compactions

Nothing was scheduled because the delta-commit threshold was not met, which on these defaults takes more delta commits than two updates produce. Setting hoodie.compact.inline.max.delta.commits to 1 at table creation went the other way: compaction then ran inline on the write itself, and the read-optimised query returned 999.99 immediately, with still nothing left for run_compaction to do. Either way the procedure returning zero rows is not a failure, and the way to verify compaction is to watch the read-optimised query catch up, not to read the procedure’s return value.

Incremental queries and time travel

This is the Hudi capability with no direct equivalent in the other formats, and it survives the trip through gRPC intact. The DataFrame writer path, where the hoodie.* options are explicit:

from decimal import Decimal
from pyspark.sql import functions as F

trips = (spark.createDataFrame(
        [(7, 107, "Kochi", Decimal("55.00"), "2026-09-06 10:00:00"),
         (8, 108, "Kochi", Decimal("65.00"), "2026-09-06 11:00:00")],
        "trip_id bigint, rider_id bigint, city string, "
        "fare decimal(10,2), started_at string")
    .withColumn("started_at", F.to_timestamp("started_at")))

options = {
    "hoodie.table.name": "trips_df",
    "hoodie.datasource.write.recordkey.field": "trip_id",
    "hoodie.datasource.write.precombine.field": "started_at",
}

(trips.write.format("hudi")
      .options(**options, **{"hoodie.datasource.write.operation": "bulk_insert"})
      .mode("overwrite").save("/tmp/warehouse/trips_df"))

The three write operations behave as their names promise, which is worth checking once rather than assuming. First an upsert of a row that already exists:

changed = (spark.createDataFrame(
        [(7, 107, "Kochi", Decimal("77.77"), "2026-09-06 12:00:00")],
        "trip_id bigint, rider_id bigint, city string, "
        "fare decimal(10,2), started_at string")
    .withColumn("started_at", F.to_timestamp("started_at")))

# upsert matches on the record key, so the row count does not grow
(changed.write.format("hudi")
        .options(**options, **{"hoodie.datasource.write.operation": "upsert"})
        .mode("append").save("/tmp/warehouse/trips_df"))

Then the same two original rows again, this time with insert:

# insert does NOT match on the record key
(trips.write.format("hudi")
      .options(**options, **{"hoodie.datasource.write.operation": "insert"})
      .mode("append").save("/tmp/warehouse/trips_df"))

after bulk_insert of 2 rows              2 rows
after upsert of trip_id 7                2 rows, fare now 77.77
after insert of the same 2 rows again    4 rows

The last line is the one to notice. insert does not deduplicate on the record key, so re-sending rows you already have doubles them. upsert is what most pipelines want, and it is not what every path defaults to.

Time travel reads a table as of an instant:

commits = [r[0] for r in spark.read.format("hudi")
           .load("/tmp/warehouse/trips_df")
           .select("_hoodie_commit_time").distinct()
           .orderBy("_hoodie_commit_time").collect()]

spark.read.format("hudi").option("as.of.instant", commits[0]) \
     .load("/tmp/warehouse/trips_df").count()

commit times            ['20260927082014648', '20260927082018250', '20260927082022472']
as.of.instant, first    2
current                 4

Two rows against four is time travel working over a gRPC connection. And the incremental read, which returns only what changed after an instant:

(spark.read.format("hudi")
      .option("hoodie.datasource.query.type", "incremental")
      .option("hoodie.datasource.read.begin.instanttime", "0")
      .load("/tmp/warehouse/trips_df")
      .count())

Passing 0 means “everything”, which is how you check the plumbing before wiring a real watermark. In production you store the last processed _hoodie_commit_time and pass that instead. The commit times are timestamps, so yours will differ from the ones above; every other number here is reproducible from a clean warehouse.

Procedures and metadata, over gRPC

People assume the CALL interface will not survive Spark Connect. It does. These all ran from the thin client:

spark.sql("CALL show_commits(table => 'default.trips_cow', limit => 10)").count()
spark.sql("CALL stats_wa(table => 'default.trips_cow')").count()
spark.sql("CALL run_clean(table => 'default.trips_cow')").count()

The table-valued functions work as well, with one argument requirement that is easy to miss:

spark.sql("SELECT count(*) FROM hudi_metadata('default.trips_cow')").collect()
spark.sql("SELECT count(*) FROM hudi_table_changes('default.trips_cow','latest_state','earliest')").collect()
spark.sql("SELECT count(*) FROM hudi_filesystem_view('default.trips_cow','city=Hyderabad')").collect()

hudi_filesystem_view takes a partition as its second argument. Called with only a table name it returns zero rows rather than an error, which reads like an empty table:

hudi_filesystem_view('default.trips_cow')                    0
hudi_filesystem_view('default.trips_cow','city=Hyderabad')   2
hudi_filesystem_view('default.trips_mor','')                 1

For a non-partitioned table pass the empty string. And hudi_table_changes in cdc mode needs a table created with CDC enabled:

IllegalArgumentException: It isn't a CDC hudi table on file:/tmp/warehouse/trips_cow

That is a table-property problem, not a Connect one. Enable hoodie.table.cdc.enabled at creation if you want the CDC feed.

rollback_to_instant only unwinds the newest instant, and not always that

This one deserves its own section, because the client-side message is the same opaque line for three different situations, and my first explanation for it was wrong. The cause is only in the Caused by: lines of the server stack trace, which the Connect client prints under JVM stacktrace:.

On a fresh three-commit table, rolling back the newest instant works:

instants = sorted(r[0] for r in
    spark.sql("CALL show_commits(table => 'default.rb_v', limit => 30)").collect())

spark.sql("CALL rollback_to_instant(table => 'default.rb_v', "
          "instant_time => '%s')" % instants[-1]).collect()
# [Row(rollback_result=True)], and the row count drops from 3 to 2

Rolling back an older instant is refused, and the cause states the rule:

UnknownException: (org.apache.hudi.exception.HoodieRollbackException)
Failed to rollback file:///tmp/warehouse/rb_u commits 20260927080539775
Caused by: org.apache.hudi.exception.HoodieRollbackException: Found commits after
time :20260927080539775, please rollback greater commits first

I first assumed a savepoint on the table was blocking rollbacks, and tested that: creating a savepoint on the second instant and then rolling back the newest one still succeeded, so the savepoint was not the cause. Rollback unwinds the tip of the timeline, one commit at a time. It is not “restore to this point”, and the way to go further back is repeated rollbacks or a savepoint plus restore.

The same call against default.trips_cow, after everything above, refused even the newest instant:

UnknownException: (org.apache.hudi.exception.HoodieRollbackException)
Failed to rollback file:///tmp/warehouse/trips_cow commits 20260927081911627
Caused by: org.apache.hudi.exception.HoodieMetadataException: Commit being rolled back
20260927081911627 is earlier than the latest compaction 20260927082049389. There are 0
deltacommits after this compaction: []

trips_cow had taken enough writes by then for Hudi’s internal metadata table to compact, and a rollback may not reach behind that compaction. The small tables had not got that far, which is why the same call worked there.

A refused rollback is not a no-op either. It leaves a pending .rollback instant on the timeline. On the three-commit table, the next plain INSERT tried to finish that pending rollback and failed with the same Found commits after time error. On trips_cow, the newest commit, which was the DROP COLUMN, was left as an inflight instant, and afterwards the table’s reported schema and its write path disagreed about whether pm existed. Run rollback_to_instant on the newest completed instant only, and check that it is the newest before you call it.

That check is where the mistake this invites lives. show_commits returns instants newest first, so the first row it hands you is the newest, while sorting those instants ascending puts the oldest first. Sorting a list of Hudi instants and passing the front of it gives rollback_to_instant exactly the argument it cannot process, which is how I produced the tip-only failure in the first place.

What the thin client cannot do

The limitations are narrow and all follow from the client not having a JVM.

spark.sparkContext

PySparkAttributeError: [JVM_ATTRIBUTE_NOT_SUPPORTED] Attribute `sparkContext` is not
supported in Spark Connect as it depends on the JVM. If you need to use this attribute,
do not use Spark Connect when creating your session. Visit
https://spark.apache.org/docs/latest/sql-getting-started.html#starting-point-sparksession
for creating regular Spark Session in detail.

spark.range(10).rdd.map(lambda r: r.id).collect()

PySparkNotImplementedError: [NOT_IMPLEMENTED] rdd is not implemented.

Neither matters much for table work, since Hudi is driven through SQL and the DataFrame API anyway. What does bite is subtler. A Python float for a decimal column fails with an empty error, where a classic session coerces it:

spark.createDataFrame(
    [(9, 109, "Kochi", 75.25, "2026-09-05 12:00:00")],
    "trip_id bigint, rider_id bigint, city string, "
    "fare decimal(10,2), started_at string")

AssertionError

That is the entire message. The traceback points at convert_decimal in pyspark/sql/conversion.py, which does assert isinstance(value, decimal.Decimal). Pass Decimal("75.25") and it works. Any Hudi table with a money column hits this on the first createDataFrame.

Common misconceptions

“One server can serve Hudi and Iceberg with no interference.” Nearly, and the exception is measurable. Loading Hudi’s session extension alongside Iceberg’s breaks Iceberg’s SQL time travel, which then fails with [INTERNAL_ERROR] Found the unresolved operator: 'TimeTravelRelation. It is one-directional: Hudi’s own DML, procedures, incremental reads and as.of.instant all keep working.

“The extension class lives under the project’s own package.” Hudi’s does not. It is org.apache.spark.sql.hudi.HoodieSparkSessionExtension, under org.apache.spark.sql. Getting it wrong does not stop the server: it starts, logs one Cannot use warning, and then every MERGE fails later for reasons that look unrelated.

”--jars is not enough for a session extension, it needs the driver classpath.” It is enough. That conclusion came from a wrong class name, not a classpath problem, and --jars alone loads both extensions correctly once the name is right.

“A bare AssertionError from the client is a client bug.” It is usually a type error. The Connect client asserts the exact type during Arrow conversion, so a Python float in a decimal column fails with an empty message where a classic session would have coerced it.

Production notes

• Treat the server config as a contract. The jar, the extension, the catalog and Kryo are fixed at startup, so a change means a restart. Nobody’s client can work around a server that was started wrong.
• Check the extension before debugging SQL. spark.conf.get("spark.sql.extensions") from the client, and grep "Cannot use" on the server log, resolve most “MERGE is not supported” reports.
• Always name columns in INSERT. It costs one line and removes the partition-reordering trap permanently, including after a later ADD COLUMNS.
• Plan for schema changes to need two configs. RENAME COLUMN, DROP COLUMN and type widening are rejected until hoodie.schema.on.read.enable and hoodie.datasource.write.schema.allow.auto.evolution.column.drop are set. Set them around the ALTER and unset them afterwards.
• Roll back only the newest completed instant. rollback_to_instant refuses anything older, refuses the newest too once the metadata table has compacted past it, and a refused rollback stays pending on the timeline.
• Pin the server’s Parquet pieces. On Spark 4.1 the bundle needs parquet-variant next to it and, for a TIMESTAMP precombine field, the GenericDataSupplier setting, or the first UPDATE fails in the write task.
• Use upsert, not insert, unless you mean to duplicate. The insert operation does not deduplicate on the record key.
• Do not judge compaction by the procedure’s return value. run_compaction returning zero rows usually means the delta-commit threshold was not met. Watch the read-optimised query instead.
• Pass Decimal for decimal columns in createDataFrame, and prefer CAST(... AS DECIMAL(10,2)) inside SQL.
• Keep Kryo set even on a server you think is Iceberg-only, because one shared server usually ends up serving both.
• Think twice before adding Hudi’s extension to a server that serves Iceberg. It breaks Iceberg’s SQL time travel, one-directionally, and nothing warns you at startup.

Frequently asked questions

Does the client need the Hudi jar? No, and that is the main reason to use Connect for Hudi work. The client sends a plan naming a table and the server resolves it. A virtualenv with pyspark is the whole client-side dependency, which also removes the jar-version skew that usually makes onboarding painful.

Can one server serve Hudi and Iceberg together? Mostly, and the cost falls entirely on Iceberg. spark.sql.extensions takes a comma-separated list, Hudi takes over spark_catalog while Iceberg gets a named catalog, and both formats’ DDL and DML work. But loading Hudi’s extension alongside Iceberg’s breaks Iceberg’s SQL time travel, which then fails with [INTERNAL_ERROR] Found the unresolved operator: 'TimeTravelRelation. Hudi is unaffected in that arrangement: its own DML, procedures, incremental reads and as.of.instant time travel all keep working. The measurement and the workaround are in Apache Iceberg through Spark Connect.

Why does my MERGE fail with a parse error? Because the session extension is not loaded on the server, from either a wrong class name or a jar missing from the server classpath. MERGE and UPDATE on Hudi are extension-provided syntax, so without it the statement is not merely unsupported, it is unparseable.

Can I set hoodie.* options per job from the client? Write-time hoodie.* options travel fine, because they are passed with the write rather than baked into the session. What you cannot change from a client is the session-level set: the extension, the catalog and the serializer.

Is the partition-column reordering a Spark Connect bug? No. It is how Hudi stores a partitioned table, and a classic session behaves the same way. Connect only makes it more visible, because the error arrives without a local stack trace to poke at.

Do client and server versions have to match? Match them. An older client was tolerated in the runs for this post: a 3.5.9 client ran DDL, INSERT, UPDATE, MERGE, a CALL and createDataFrame against the 4.1.3 Hudi server. A newer one is not: a 4.1.3 client against a 3.5.9 server failed on createDataFrame with SQL_CONF_NOT_FOUND for spark.sql.session.localRelationChunkSizeRows, because it expects a config the older server does not have. That is why the install line above is pinned.

Conclusion

The useful reframing is that Spark Connect forces a separation which was always there. Hudi is infrastructure: a jar, a Catalyst extension, a catalog and a serializer. Your query is not. Classic Spark let you blur the two by shipping jars with every submit, and most --packages lines in most runbooks are that blur made permanent.

Connect makes you decide once, on the server, and then hands every client a clean API with nothing to install. The cost is exactly the flexibility you were abusing: no per-job jars, no per-job extensions, no sparkContext escape hatch. For a team sharing a Hudi lakehouse, that is a good trade.

What is worth carrying away is how few of the failures were about Spark Connect. An extension class under a package that looks wrong but is not, a partition column silently moved to the end of the schema, insert where upsert was meant, rollback_to_instant that only unwinds the tip, and a float where a Decimal belongs. Every one of those behaves identically in a classic session. Reproduce a failure locally before blaming the transport, which is the habit that turned two of my own confident wrong answers here into the right ones.

References

• Spark Connect overview for the architecture and the client APIs
• Hudi Spark SQL DDL and SQL procedures for table properties and the CALL surface
• Hudi configuration reference for the hoodie.* write and read options used above
• conversion.py, whose convert_decimal is the assertion behind the empty AssertionError
• Apache Hudi architecture for the timeline, file slices and compaction machinery underneath all of this

Trademarks

Apache Spark, Apache Hudi, Apache Iceberg, Apache Hive, Apache Parquet and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.