This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 3015eb944083 test(integ-test): add Testcontainers E2E for
VECTOR/BLOB/VARIANT Hive… (#19203)
3015eb944083 is described below
commit 3015eb944083272459040dc5da5755e683674cb5
Author: voonhous <[email protected]>
AuthorDate: Wed Jul 8 18:57:59 2026 +0800
test(integ-test): add Testcontainers E2E for VECTOR/BLOB/VARIANT Hive…
(#19203)
* test(integ-test): add Testcontainers E2E for VECTOR/BLOB/VARIANT Hive sync
Testcontainers-based E2E coverage that a Spark write of Hudi's custom
logical types produces a Hive-syncable schema queryable through a real
Hive metastore. ITTestCustomTypeHiveSync exercises VECTOR and BLOB (SQL +
DataFrame) plus Spark 4.0 VARIANT (Spark4-gated), asserting write ->
hive-sync -> Hive DESCRIBE/SHOW PARTITIONS/SELECT round-trips.
Runs on both a Spark 3.5.3 (JDK 11) and Spark 4.0.2 (JDK 17) stack against
Hive 2.3.10, via a new integration-tests-hive-sync CI job. The integ2
package manages its own docker stack and is excluded from the legacy
in-job compose demo to avoid host-port collisions.
Scoped decomposition of #18535 (Hive-sync half only): the Trino read E2E
and two bundled production features (blob partial-struct write, Lance
VARIANT support) are deferred to follow-up PRs.
* test(integ-test): address review comments on testcontainers harness
- delete unused-and-broken HiveService.runQuery (execInContainer has no
shell, so the added literal quotes would reach hive -e and break parsing)
- fix assertStdOutContains javadoc: asserts exactly once, not at least once
- gate integration-tests-hive-sync CI job on the changes job's relevance
output, matching the legacy integration-tests job
- trim unused harness API: sparkAdhoc2, unused Containers/Paths/Network
constants,
CommandResult.expectToFail/assertExitCodeIs/assertStdErrContains,
HiveService.executeFile/copyFile, SparkService.copyFile,
CommandExecutor.copyFileToContainer
- register compose temp files for deleteOnExit so the temp dir actually
deletes; add EOF newline; fix ContainerProvider javadoc; rely on root
dependencyManagement for the managed testcontainers artifact
* test(integ-test): use jdbc hive_sync mode for VARIANT SQL fixture
The VARIANT SQL fixture was the only one syncing via hms mode while
retaining a jdbcurl (used only by jdbc mode). Align it with every other
fixture, including the DataFrame VARIANT fixture that syncs the identical
VariantType, which all use jdbc.
* test(integ-test): move VERBOSE_HIVECONFS into SystemProps
Keep all verbose-mode config in one place. VERBOSE_HIVECONFS was the only
constant at the outer-class level; group it with SystemProps.HIVE_VERBOSE,
which its javadoc already references.
---
.github/workflows/bot.yml | 76 ++++++
...r-compose_hadoop284_hive2310_spark353_amd64.yml | 76 ------
...r-compose_hadoop284_hive2310_spark353_arm64.yml | 72 ------
docker/demo/sparksql-blob-type-df.commands | 108 +++++++++
docker/demo/sparksql-blob-type-sql.commands | 126 ++++++++++
docker/demo/sparksql-variant-type-df.commands | 80 +++++++
docker/demo/sparksql-variant-type-sql.commands | 89 +++++++
docker/demo/sparksql-vector-type-df.commands | 90 +++++++
docker/demo/sparksql-vector-type-sql.commands | 90 +++++++
docker/hoodie/hadoop/hive_base/entrypoint.sh | 2 +-
docker/hoodie/hadoop/sparkmaster/master.sh | 2 +-
hudi-integ-test/pom.xml | 48 +++-
.../integ2/testcontainers/ContainerProvider.java | 37 +++
.../testcontainers/ITTestBaseTestcontainers.java | 236 ++++++++++++++++++
.../testcontainers/ITTestCustomTypeHiveSync.java | 266 +++++++++++++++++++++
.../testcontainers/TestcontainersConfig.java | 110 +++++++++
.../testcontainers/command/CommandExecutor.java | 87 +++++++
.../testcontainers/command/CommandResult.java | 137 +++++++++++
.../integ2/testcontainers/service/HiveService.java | 72 ++++++
.../testcontainers/service/SparkService.java | 71 ++++++
20 files changed, 1724 insertions(+), 151 deletions(-)
diff --git a/.github/workflows/bot.yml b/.github/workflows/bot.yml
index 99fb6ba9a8be..c6901e00503c 100644
--- a/.github/workflows/bot.yml
+++ b/.github/workflows/bot.yml
@@ -1432,6 +1432,82 @@ jobs:
flags: integration-tests
token: ${{ secrets.CODECOV_TOKEN }}
+ integration-tests-hive-sync:
+ # Testcontainers-based E2E hive sync coverage for Hudi's custom logical
types
+ # (VECTOR, BLOB) and the Spark 4.0 VARIANT type. Runs the integ2
testcontainers
+ # suite (ITTestCustomTypeHiveSync) against a real Hive metastore on both a
+ # Spark 3.5.3 and Spark 4.0.2 stack.
+ runs-on: ubuntu-latest
+ needs: changes
+ strategy:
+ fail-fast: false
+ matrix:
+ include:
+ - sparkProfile: 'spark3.5'
+ scalaProfile: '-Dscala-2.12 -Dscala.binary.version=2.12'
+ flinkProfile: 'flink1.20'
+ jdkVersion: '11'
+ composePrefix: 'docker-compose_hadoop284_hive2310_spark353'
+ sparkAdhocImage:
'apachehudi/hudi-hadoop_2.8.4-hive_2.3.10-sparkadhoc_3.5.3:latest'
+ - sparkProfile: 'spark4.0'
+ scalaProfile: '-Dscala-2.13 -Dscala.binary.version=2.13'
+ flinkProfile: 'flink1.20'
+ jdkVersion: '17'
+ composePrefix: 'docker-compose_hadoop340_hive2310_spark402'
+ sparkAdhocImage:
'apachehudi/hudi-hadoop_3.4.0-hive_2.3.10-sparkadhoc_4.0.2:latest'
+ steps:
+ - if: needs.changes.outputs.relevant == 'true'
+ uses: actions/checkout@v5
+ - name: Set up JDK ${{ matrix.jdkVersion }}
+ if: needs.changes.outputs.relevant == 'true'
+ uses: actions/setup-java@v5
+ with:
+ java-version: ${{ matrix.jdkVersion }}
+ distribution: 'temurin'
+ architecture: x64
+ cache: maven
+ - name: Free disk space
+ if: needs.changes.outputs.relevant == 'true'
+ run: |
+ sudo rm -rf /usr/share/dotnet
+ sudo rm -rf /usr/local/lib/android
+ sudo rm -rf /opt/ghc
+ sudo rm -rf /usr/local/share/boost
+ docker system prune --all --force --volumes
+ - name: Pre-pull compose images (fails fast if not published)
+ if: needs.changes.outputs.relevant == 'true'
+ env:
+ SPARK_ADHOC_IMAGE: ${{ matrix.sparkAdhocImage }}
+ run: |
+ # Surface missing Spark 4.0.2 images before the 15-minute Maven
install.
+ # The remaining images in the compose stack are pulled by
docker-compose at
+ # test time.
+ docker pull "$SPARK_ADHOC_IMAGE"
+ - name: Build and install Hudi artifacts
+ if: needs.changes.outputs.relevant == 'true'
+ env:
+ SPARK_PROFILE: ${{ matrix.sparkProfile }}
+ FLINK_PROFILE: ${{ matrix.flinkProfile }}
+ SCALA_PROFILE: ${{ matrix.scalaProfile }}
+ run:
+ mvn clean install -T 2 $SCALA_PROFILE -D"$SPARK_PROFILE"
-D"$FLINK_PROFILE" -Pintegration-tests -DskipTests=true
-Ddocker.compose.skip=true $MVN_ARGS
+ - name: Run integ2 testcontainers suite
+ if: needs.changes.outputs.relevant == 'true'
+ env:
+ SPARK_PROFILE: ${{ matrix.sparkProfile }}
+ SCALA_PROFILE: ${{ matrix.scalaProfile }}
+ COMPOSE_PREFIX: ${{ matrix.composePrefix }}
+ run: |
+ # -DskipITs=false overrides the spark4.0 profile's skipITs=true
default
+ # (see root pom.xml). Without it, failsafe skips all ITs on the
spark4.0 matrix row.
+ mvn verify $SCALA_PROFILE -D"$SPARK_PROFILE" -Pintegration-tests \
+ -pl hudi-integ-test \
+ -DskipITs=false \
+ -Ddocker.compose.skip=true \
+ -Dit.test='ITTestCustomTypeHiveSync' \
+ -Dspark.docker.compose.prefix=$COMPOSE_PREFIX \
+ $MVN_ARGS
+
build-spark-java17:
runs-on: ubuntu-latest
needs: changes
diff --git
a/docker/compose/docker-compose_hadoop284_hive2310_spark353_amd64.yml
b/docker/compose/docker-compose_hadoop284_hive2310_spark353_amd64.yml
index 1d84417b378a..c302a3c95cca 100644
--- a/docker/compose/docker-compose_hadoop284_hive2310_spark353_amd64.yml
+++ b/docker/compose/docker-compose_hadoop284_hive2310_spark353_amd64.yml
@@ -186,78 +186,6 @@ services:
- KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
- ALLOW_PLAINTEXT_LISTENER=yes
- presto-coordinator-1:
- container_name: presto-coordinator-1
- hostname: presto-coordinator-1
- image: apachehudi/hudi-hadoop_2.8.4-prestobase_0.271:latest
- platform: linux/amd64
- ports:
- - "8090:8090"
- environment:
- - PRESTO_JVM_MAX_HEAP=512M
- - PRESTO_QUERY_MAX_MEMORY=1GB
- - PRESTO_QUERY_MAX_MEMORY_PER_NODE=256MB
- - PRESTO_QUERY_MAX_TOTAL_MEMORY_PER_NODE=384MB
- - PRESTO_MEMORY_HEAP_HEADROOM_PER_NODE=100MB
- - TERM=xterm
- links:
- - "hivemetastore"
- volumes:
- - ${HUDI_WS}:/var/hoodie/ws
- command: coordinator
-
- presto-worker-1:
- container_name: presto-worker-1
- hostname: presto-worker-1
- image: apachehudi/hudi-hadoop_2.8.4-prestobase_0.271:latest
- platform: linux/amd64
- depends_on: [ "presto-coordinator-1" ]
- environment:
- - PRESTO_JVM_MAX_HEAP=512M
- - PRESTO_QUERY_MAX_MEMORY=1GB
- - PRESTO_QUERY_MAX_MEMORY_PER_NODE=256MB
- - PRESTO_QUERY_MAX_TOTAL_MEMORY_PER_NODE=384MB
- - PRESTO_MEMORY_HEAP_HEADROOM_PER_NODE=100MB
- - TERM=xterm
- links:
- - "hivemetastore"
- - "hiveserver"
- - "hive-metastore-postgresql"
- - "namenode"
- volumes:
- - ${HUDI_WS}:/var/hoodie/ws
- command: worker
-
- trino-coordinator-1:
- container_name: trino-coordinator-1
- hostname: trino-coordinator-1
- image: apachehudi/hudi-hadoop_2.8.4-trinocoordinator_368:latest
- platform: linux/amd64
- ports:
- - "8091:8091"
- links:
- - "hivemetastore"
- volumes:
- - ${HUDI_WS}:/var/hoodie/ws
- command: http://trino-coordinator-1:8091 trino-coordinator-1
-
- trino-worker-1:
- container_name: trino-worker-1
- hostname: trino-worker-1
- image: apachehudi/hudi-hadoop_2.8.4-trinoworker_368:latest
- platform: linux/amd64
- depends_on: [ "trino-coordinator-1" ]
- ports:
- - "8092:8092"
- links:
- - "hivemetastore"
- - "hiveserver"
- - "hive-metastore-postgresql"
- - "namenode"
- volumes:
- - ${HUDI_WS}:/var/hoodie/ws
- command: http://trino-coordinator-1:8091 trino-worker-1
-
graphite:
container_name: graphite
hostname: graphite
@@ -287,8 +215,6 @@ services:
- "hiveserver"
- "hive-metastore-postgresql"
- "namenode"
- - "presto-coordinator-1"
- - "trino-coordinator-1"
volumes:
- ${HUDI_WS}:/var/hoodie/ws
@@ -311,8 +237,6 @@ services:
- "hiveserver"
- "hive-metastore-postgresql"
- "namenode"
- - "presto-coordinator-1"
- - "trino-coordinator-1"
volumes:
- ${HUDI_WS}:/var/hoodie/ws
diff --git
a/docker/compose/docker-compose_hadoop284_hive2310_spark353_arm64.yml
b/docker/compose/docker-compose_hadoop284_hive2310_spark353_arm64.yml
index b995859c8022..a9f042155d30 100644
--- a/docker/compose/docker-compose_hadoop284_hive2310_spark353_arm64.yml
+++ b/docker/compose/docker-compose_hadoop284_hive2310_spark353_arm64.yml
@@ -179,74 +179,6 @@ services:
- KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
- ALLOW_PLAINTEXT_LISTENER=yes
- presto-coordinator-1:
- container_name: presto-coordinator-1
- hostname: presto-coordinator-1
- image: apachehudi/hudi-hadoop_2.8.4-prestobase_0.271:latest
- ports:
- - "8090:8090"
- environment:
- - PRESTO_JVM_MAX_HEAP=512M
- - PRESTO_QUERY_MAX_MEMORY=1GB
- - PRESTO_QUERY_MAX_MEMORY_PER_NODE=256MB
- - PRESTO_QUERY_MAX_TOTAL_MEMORY_PER_NODE=384MB
- - PRESTO_MEMORY_HEAP_HEADROOM_PER_NODE=100MB
- - TERM=xterm
- links:
- - "hivemetastore"
- volumes:
- - ${HUDI_WS}:/var/hoodie/ws
- command: coordinator
-
- presto-worker-1:
- container_name: presto-worker-1
- hostname: presto-worker-1
- image: apachehudi/hudi-hadoop_2.8.4-prestobase_0.271:latest
- depends_on: [ "presto-coordinator-1" ]
- environment:
- - PRESTO_JVM_MAX_HEAP=512M
- - PRESTO_QUERY_MAX_MEMORY=1GB
- - PRESTO_QUERY_MAX_MEMORY_PER_NODE=256MB
- - PRESTO_QUERY_MAX_TOTAL_MEMORY_PER_NODE=384MB
- - PRESTO_MEMORY_HEAP_HEADROOM_PER_NODE=100MB
- - TERM=xterm
- links:
- - "hivemetastore"
- - "hiveserver"
- - "hive-metastore-postgresql"
- - "namenode"
- volumes:
- - ${HUDI_WS}:/var/hoodie/ws
- command: worker
-
- trino-coordinator-1:
- container_name: trino-coordinator-1
- hostname: trino-coordinator-1
- image: apachehudi/hudi-hadoop_2.8.4-trinocoordinator_368:latest
- ports:
- - "8091:8091"
- links:
- - "hivemetastore"
- volumes:
- - ${HUDI_WS}:/var/hoodie/ws
- command: http://trino-coordinator-1:8091 trino-coordinator-1
-
- trino-worker-1:
- container_name: trino-worker-1
- hostname: trino-worker-1
- image: apachehudi/hudi-hadoop_2.8.4-trinoworker_368:latest
- depends_on: [ "trino-coordinator-1" ]
- ports:
- - "8092:8092"
- links:
- - "hivemetastore"
- - "hiveserver"
- - "hive-metastore-postgresql"
- - "namenode"
- volumes:
- - ${HUDI_WS}:/var/hoodie/ws
- command: http://trino-coordinator-1:8091 trino-worker-1
-
graphite:
container_name: graphite
hostname: graphite
@@ -275,8 +207,6 @@ services:
- "hiveserver"
- "hive-metastore-postgresql"
- "namenode"
- - "presto-coordinator-1"
- - "trino-coordinator-1"
volumes:
- ${HUDI_WS}:/var/hoodie/ws
@@ -298,8 +228,6 @@ services:
- "hiveserver"
- "hive-metastore-postgresql"
- "namenode"
- - "presto-coordinator-1"
- - "trino-coordinator-1"
volumes:
- ${HUDI_WS}:/var/hoodie/ws
diff --git a/docker/demo/sparksql-blob-type-df.commands
b/docker/demo/sparksql-blob-type-df.commands
new file mode 100644
index 000000000000..28e0663d74e9
--- /dev/null
+++ b/docker/demo/sparksql-blob-type-df.commands
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.{Row, SaveMode}
+
+// BLOB is a struct with fields: type (string), data (binary, nullable),
reference (struct, nullable)
+val blobMetadata = new MetadataBuilder().putString("hudi_type", "BLOB").build()
+
+val referenceType = StructType(Seq(
+ StructField("external_path", StringType, nullable = false),
+ StructField("offset", LongType, nullable = true),
+ StructField("length", LongType, nullable = true),
+ StructField("managed", BooleanType, nullable = false)
+))
+
+val blobType = StructType(Seq(
+ StructField("type", StringType, nullable = false),
+ StructField("data", BinaryType, nullable = true),
+ StructField("reference", referenceType, nullable = true)
+))
+
+val schema = StructType(Seq(
+ StructField("id", LongType, nullable = false),
+ StructField("name", StringType),
+ StructField("blob_data", blobType, nullable = false, metadata =
blobMetadata),
+ StructField("dt", StringType)
+))
+
+// Shared Hive-sync + write options factored out so the upsert and delete
writes
+// reuse the exact same sync configuration as the initial Overwrite.
+// Body is wrapped in { } so the spark-shell REPL keeps the chained
.option(...)
+// calls attached to the def. Without braces, writer.format("hudi") on the
first
+// body line parses as a complete expression, the REPL closes the def there,
and
+// the remaining .option(...) lines get dot-applied to `sc` (SparkContext).
+def applyWriteOpts(writer: org.apache.spark.sql.DataFrameWriter[Row]):
org.apache.spark.sql.DataFrameWriter[Row] = {
+ writer.format("hudi")
+ .option("hoodie.table.name", "blob_test_df")
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.precombine.field", "name")
+ .option("hoodie.datasource.write.partitionpath.field", "dt")
+ .option("hoodie.datasource.hive_sync.enable", "true")
+ .option("hoodie.datasource.hive_sync.database", "default")
+ .option("hoodie.datasource.hive_sync.table", "blob_test_df")
+ .option("hoodie.datasource.hive_sync.jdbcurl",
"jdbc:hive2://hiveserver:10000/")
+ .option("hoodie.datasource.hive_sync.mode", "jdbc")
+ .option("hoodie.datasource.hive_sync.partition_fields", "dt")
+ .option("hoodie.datasource.hive_sync.partition_extractor_class",
+ "org.apache.hudi.hive.MultiPartKeysValueExtractor")
+ .option("hoodie.datasource.hive_sync.username", "hive")
+ .option("hoodie.datasource.hive_sync.password", "hive")
+}
+
+// Seed two rows in dt='2024-01-01' via Overwrite.
+val seed = Seq(
+ Row(1L, "file1", Row("INLINE", "hello world".getBytes, null), "2024-01-01"),
+ Row(2L, "file2", Row("INLINE", "test data".getBytes, null), "2024-01-01")
+)
+val seedDf = spark.createDataFrame(spark.sparkContext.parallelize(seed),
schema)
+applyWriteOpts(seedDf.write).mode(SaveMode.Overwrite).save("/user/hive/warehouse/blob_test_df")
+spark.sql("select id, name, blob_data.type, dt from blob_test_df order by
id").show(10, false)
+println("BLOB_DF_INSERT_SUCCESS")
+
+// Append-upsert: id=2 value changes, id=3 new in a new partition
dt='2024-01-02'.
+// precombine=name must strictly increase for the matched row so Hudi keeps the
+// new value (alphabetical 'file2-updated' > 'file2').
+val upsertRows = Seq(
+ Row(2L, "file2-updated", Row("INLINE", "updated payload".getBytes, null),
"2024-01-01"),
+ Row(3L, "file3", Row("INLINE", "brand new".getBytes, null), "2024-01-02")
+)
+val upsertDf =
spark.createDataFrame(spark.sparkContext.parallelize(upsertRows), schema)
+applyWriteOpts(upsertDf.write)
+ .option("hoodie.datasource.write.operation", "upsert")
+ .mode(SaveMode.Append)
+ .save("/user/hive/warehouse/blob_test_df")
+spark.sql("select id, name, blob_data.type, dt from blob_test_df order by
id").show(10, false)
+println("BLOB_DF_UPSERT_SUCCESS")
+
+// Delete id=3 via operation=delete. The DF still needs a schema-compatible
BLOB
+// column (Hudi uses only the record key/partition for delete, the payload is
+// ignored but must deserialize).
+val deleteRows = Seq(
+ Row(3L, "file3", Row("INLINE", "ignored".getBytes, null), "2024-01-02")
+)
+val deleteDf =
spark.createDataFrame(spark.sparkContext.parallelize(deleteRows), schema)
+applyWriteOpts(deleteDf.write)
+ .option("hoodie.datasource.write.operation", "delete")
+ .mode(SaveMode.Append)
+ .save("/user/hive/warehouse/blob_test_df")
+spark.sql("select id, name, blob_data.type, dt from blob_test_df order by
id").show(10, false)
+println("BLOB_DF_DELETE_SUCCESS")
+
+println("BLOB_DF_TEST_SUCCESS")
diff --git a/docker/demo/sparksql-blob-type-sql.commands
b/docker/demo/sparksql-blob-type-sql.commands
new file mode 100644
index 000000000000..8a03b1d16f30
--- /dev/null
+++ b/docker/demo/sparksql-blob-type-sql.commands
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// E2E test: BLOB type with Hive sync via SQL CREATE TABLE path.
+// BLOB is parsed by HoodieSpark3_5ExtendedSqlAstBuilder into a struct
+// <type:string, data:binary, reference:struct<...>>.
+spark.sql("""
+ CREATE TABLE blob_test (
+ id LONG,
+ name STRING,
+ blob_data BLOB,
+ dt STRING
+ ) USING hudi
+ PARTITIONED BY (dt)
+ LOCATION '/user/hive/warehouse/blob_test'
+ TBLPROPERTIES (
+ 'primaryKey' = 'id',
+ 'preCombineField' = 'name',
+ 'hoodie.datasource.hive_sync.enable' = 'true',
+ 'hoodie.datasource.hive_sync.database' = 'default',
+ 'hoodie.datasource.hive_sync.table' = 'blob_test',
+ 'hoodie.datasource.hive_sync.jdbcurl' = 'jdbc:hive2://hiveserver:10000/',
+ 'hoodie.datasource.hive_sync.mode' = 'jdbc',
+ 'hoodie.datasource.hive_sync.partition_fields' = 'dt',
+ 'hoodie.datasource.hive_sync.partition_extractor_class' =
'org.apache.hudi.hive.MultiPartKeysValueExtractor',
+ 'hoodie.datasource.hive_sync.username' = 'hive',
+ 'hoodie.datasource.hive_sync.password' = 'hive'
+ )
+""")
+
+// Seed two rows in dt='2024-01-01'. Use OUT_OF_LINE here so subsequent
UPDATE/MERGE
+// can mutate to a different reference without changing the struct shape
between ops
+// (avoids mixing INLINE and OUT_OF_LINE branches in a single test).
+spark.sql("""
+ INSERT INTO blob_test VALUES
+ (1, 'file1', named_struct(
+ 'type', 'OUT_OF_LINE',
+ 'data', cast(null as binary),
+ 'reference', named_struct(
+ 'external_path', 'blobs/seed-1',
+ 'offset', 0L,
+ 'length', 11L,
+ 'managed', false)), '2024-01-01'),
+ (2, 'file2', named_struct(
+ 'type', 'OUT_OF_LINE',
+ 'data', cast(null as binary),
+ 'reference', named_struct(
+ 'external_path', 'blobs/seed-2',
+ 'offset', 0L,
+ 'length', 9L,
+ 'managed', false)), '2024-01-01')
+""")
+spark.sql("select id, name, blob_data.type, dt from blob_test").show(10, false)
+println("BLOB_SQL_INSERT_SUCCESS")
+
+// UPDATE exercises the BLOB metadata re-attach on the UPDATE write path.
+// Per RFC-100 external_path and managed are non-null.
+spark.sql("""
+ UPDATE blob_test
+ SET blob_data = named_struct(
+ 'type', 'OUT_OF_LINE',
+ 'data', cast(null as binary),
+ 'reference', named_struct(
+ 'external_path', 'blobs/updated-1',
+ 'offset', 10L,
+ 'length', 100L,
+ 'managed', true)),
+ name = 'file1-updated'
+ WHERE id = 1
+""")
+spark.sql("select id, name, blob_data.reference.external_path from blob_test
where id = 1").show(10, false)
+println("BLOB_SQL_UPDATE_SUCCESS")
+
+// MERGE exercises both MATCHED (UPDATE SET) and NOT MATCHED (INSERT of a new
+// row into a new partition dt='2024-01-02'). The USING CTE strips the
hudi_type
+// metadata from the struct column, so this specifically validates the reattach
+// path for MERGE.
+spark.sql("""
+ MERGE INTO blob_test t
+ USING (
+ SELECT 2L AS id, 'file2-merged' AS name, named_struct(
+ 'type', 'OUT_OF_LINE',
+ 'data', cast(null as binary),
+ 'reference', named_struct(
+ 'external_path', 'blobs/merged-2',
+ 'offset', 20L,
+ 'length', 200L,
+ 'managed', true)) AS blob_data, '2024-01-01' AS dt
+ UNION ALL
+ SELECT 3L AS id, 'file3' AS name, named_struct(
+ 'type', 'OUT_OF_LINE',
+ 'data', cast(null as binary),
+ 'reference', named_struct(
+ 'external_path', 'blobs/inserted-3',
+ 'offset', 300L,
+ 'length', 30L,
+ 'managed', false)) AS blob_data, '2024-01-02' AS dt
+ ) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET t.name = s.name, t.blob_data = s.blob_data
+ WHEN NOT MATCHED THEN INSERT (id, name, blob_data, dt) VALUES (s.id, s.name,
s.blob_data, s.dt)
+""")
+spark.sql("select id, name, blob_data.reference.external_path, dt from
blob_test order by id").show(10, false)
+println("BLOB_SQL_MERGE_SUCCESS")
+
+// DELETE the NOT MATCHED row we just inserted, returning to 2 rows total.
+spark.sql("DELETE FROM blob_test WHERE id = 3")
+spark.sql("select id, name, blob_data.type, dt from blob_test order by
id").show(10, false)
+println("BLOB_SQL_DELETE_SUCCESS")
+
+println("BLOB_SQL_TEST_SUCCESS")
diff --git a/docker/demo/sparksql-variant-type-df.commands
b/docker/demo/sparksql-variant-type-df.commands
new file mode 100644
index 000000000000..dbbeabdad039
--- /dev/null
+++ b/docker/demo/sparksql-variant-type-df.commands
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import org.apache.spark.sql.{DataFrame, SaveMode}
+
+// E2E test: VARIANT type with Hive sync via DataFrame API path.
+// Tests that the HiveSyncTool correctly maps VARIANT to
struct<value:binary,metadata:binary> in Hive.
+
+// Body is wrapped in { } so the spark-shell REPL keeps the chained
.option(...)
+// calls attached to the def. Without braces, writer.format("hudi") on the
first
+// body line parses as a complete expression, the REPL closes the def there,
and
+// the remaining .option(...) lines get dot-applied to `sc` (SparkContext).
+def applyWriteOpts(writer: org.apache.spark.sql.DataFrameWriter[_]):
org.apache.spark.sql.DataFrameWriter[_] = {
+ writer.format("hudi")
+ .option("hoodie.table.name", "variant_test_df")
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.precombine.field", "name")
+ .option("hoodie.datasource.write.partitionpath.field", "dt")
+ .option("hoodie.datasource.hive_sync.enable", "true")
+ .option("hoodie.datasource.hive_sync.database", "default")
+ .option("hoodie.datasource.hive_sync.table", "variant_test_df")
+ .option("hoodie.datasource.hive_sync.jdbcurl",
"jdbc:hive2://hiveserver:10000/")
+ .option("hoodie.datasource.hive_sync.mode", "jdbc")
+ .option("hoodie.datasource.hive_sync.partition_fields", "dt")
+ .option("hoodie.datasource.hive_sync.partition_extractor_class",
+ "org.apache.hudi.hive.MultiPartKeysValueExtractor")
+ .option("hoodie.datasource.hive_sync.username", "hive")
+ .option("hoodie.datasource.hive_sync.password", "hive")
+}
+
+// Seed two rows in dt='2024-01-01' via Overwrite.
+val seedDf: DataFrame = spark.sql("""
+ SELECT 1L AS id, 'row1' AS name, parse_json('{"key":"value1"}') AS
variant_data, '2024-01-01' AS dt
+ UNION ALL
+ SELECT 2L AS id, 'row2' AS name, parse_json('{"key":"value2"}') AS
variant_data, '2024-01-01' AS dt
+""")
+applyWriteOpts(seedDf.write).mode(SaveMode.Overwrite).save("/user/hive/warehouse/variant_test_df")
+spark.sql("select id, name, cast(variant_data as string), dt from
variant_test_df order by id").show(10, false)
+println("VARIANT_DF_INSERT_SUCCESS")
+
+// Append-upsert: id=2 mutated, id=3 new in dt='2024-01-02'.
+val upsertDf: DataFrame = spark.sql("""
+ SELECT 2L AS id, 'row2-updated' AS name,
parse_json('{"key":"value2-updated"}') AS variant_data, '2024-01-01' AS dt
+ UNION ALL
+ SELECT 3L AS id, 'row3' AS name, parse_json('{"key":"value3"}') AS
variant_data, '2024-01-02' AS dt
+""")
+applyWriteOpts(upsertDf.write)
+ .option("hoodie.datasource.write.operation", "upsert")
+ .mode(SaveMode.Append)
+ .save("/user/hive/warehouse/variant_test_df")
+spark.sql("select id, name, cast(variant_data as string), dt from
variant_test_df order by id").show(10, false)
+println("VARIANT_DF_UPSERT_SUCCESS")
+
+// Delete id=3 via operation=delete.
+val deleteDf: DataFrame = spark.sql("""
+ SELECT 3L AS id, 'row3' AS name, parse_json('{}') AS variant_data,
'2024-01-02' AS dt
+""")
+applyWriteOpts(deleteDf.write)
+ .option("hoodie.datasource.write.operation", "delete")
+ .mode(SaveMode.Append)
+ .save("/user/hive/warehouse/variant_test_df")
+spark.sql("select id, name, cast(variant_data as string), dt from
variant_test_df order by id").show(10, false)
+println("VARIANT_DF_DELETE_SUCCESS")
+
+println("VARIANT_DF_TEST_SUCCESS")
diff --git a/docker/demo/sparksql-variant-type-sql.commands
b/docker/demo/sparksql-variant-type-sql.commands
new file mode 100644
index 000000000000..c21a62674ef3
--- /dev/null
+++ b/docker/demo/sparksql-variant-type-sql.commands
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// E2E test: VARIANT type with Hive sync via SQL CREATE TABLE path.
+// Tests that VariantType is mapped to struct<value:binary,metadata:binary> in
Hive.
+spark.sql("""
+ CREATE TABLE variant_test (
+ id LONG,
+ name STRING,
+ variant_data VARIANT,
+ dt STRING
+ ) USING hudi
+ PARTITIONED BY (dt)
+ LOCATION '/user/hive/warehouse/variant_test'
+ TBLPROPERTIES (
+ 'primaryKey' = 'id',
+ 'preCombineField' = 'name',
+ 'hoodie.datasource.hive_sync.enable' = 'true',
+ 'hoodie.datasource.hive_sync.database' = 'default',
+ 'hoodie.datasource.hive_sync.table' = 'variant_test',
+ 'hoodie.datasource.hive_sync.jdbcurl' = 'jdbc:hive2://hiveserver:10000/',
+ 'hoodie.datasource.hive_sync.mode' = 'jdbc',
+ 'hoodie.datasource.hive_sync.partition_fields' = 'dt',
+ 'hoodie.datasource.hive_sync.partition_extractor_class' =
'org.apache.hudi.hive.MultiPartKeysValueExtractor',
+ 'hoodie.datasource.hive_sync.username' = 'hive',
+ 'hoodie.datasource.hive_sync.password' = 'hive'
+ )
+""")
+
+spark.sql("""
+ INSERT INTO variant_test VALUES
+ (1, 'row1', parse_json('{"key":"value1"}'), '2024-01-01'),
+ (2, 'row2', parse_json('{"key":"value2"}'), '2024-01-01')
+""")
+spark.sql("select id, name, cast(variant_data as string), dt from variant_test
order by id").show(10, false)
+println("VARIANT_SQL_INSERT_SUCCESS")
+
+// UPDATE against a VariantType column. Hudi's V1 writer accepts VariantType
+// writes on Spark 4 (see fix f395eea4b4a9); UPDATE exercises the same writer
+// path with castIfNeeded applied to the assignment.
+spark.sql("""
+ UPDATE variant_test
+ SET variant_data = parse_json('{"key":"value1-updated"}'),
+ name = 'row1-updated'
+ WHERE id = 1
+""")
+spark.sql("select id, name, cast(variant_data as string) from variant_test
where id = 1").show(10, false)
+println("VARIANT_SQL_UPDATE_SUCCESS")
+
+// MERGE exercises both MATCHED (UPDATE) and NOT MATCHED (INSERT into new
+// partition dt='2024-01-02').
+spark.sql("""
+ MERGE INTO variant_test t
+ USING (
+ SELECT 2L AS id, 'row2-merged' AS name,
+ parse_json('{"key":"value2-merged"}') AS variant_data,
+ '2024-01-01' AS dt
+ UNION ALL
+ SELECT 3L AS id, 'row3' AS name,
+ parse_json('{"key":"value3"}') AS variant_data,
+ '2024-01-02' AS dt
+ ) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET t.name = s.name, t.variant_data = s.variant_data
+ WHEN NOT MATCHED THEN INSERT (id, name, variant_data, dt) VALUES (s.id,
s.name, s.variant_data, s.dt)
+""")
+spark.sql("select id, name, cast(variant_data as string), dt from variant_test
order by id").show(10, false)
+println("VARIANT_SQL_MERGE_SUCCESS")
+
+spark.sql("DELETE FROM variant_test WHERE id = 3")
+spark.sql("select id, name, cast(variant_data as string), dt from variant_test
order by id").show(10, false)
+println("VARIANT_SQL_DELETE_SUCCESS")
+
+println("VARIANT_SQL_TEST_SUCCESS")
diff --git a/docker/demo/sparksql-vector-type-df.commands
b/docker/demo/sparksql-vector-type-df.commands
new file mode 100644
index 000000000000..ef71771be295
--- /dev/null
+++ b/docker/demo/sparksql-vector-type-df.commands
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.{Row, SaveMode}
+
+// Create schema with VECTOR metadata + partition column
+val metadata = new MetadataBuilder().putString("hudi_type",
"VECTOR(3)").build()
+val schema = StructType(Seq(
+ StructField("id", LongType, nullable = false),
+ StructField("name", StringType),
+ StructField("embedding", ArrayType(FloatType, containsNull = false),
nullable = false, metadata = metadata),
+ StructField("dt", StringType)
+))
+
+// Body is wrapped in { } so the spark-shell REPL keeps the chained
.option(...)
+// calls attached to the def. Without braces, writer.format("hudi") on the
first
+// body line parses as a complete expression, the REPL closes the def there,
and
+// the remaining .option(...) lines get dot-applied to `sc` (SparkContext).
+def applyWriteOpts(writer: org.apache.spark.sql.DataFrameWriter[Row]):
org.apache.spark.sql.DataFrameWriter[Row] = {
+ writer.format("hudi")
+ .option("hoodie.table.name", "vector_test_df")
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.precombine.field", "name")
+ .option("hoodie.datasource.write.partitionpath.field", "dt")
+ .option("hoodie.datasource.hive_sync.enable", "true")
+ .option("hoodie.datasource.hive_sync.database", "default")
+ .option("hoodie.datasource.hive_sync.table", "vector_test_df")
+ .option("hoodie.datasource.hive_sync.jdbcurl",
"jdbc:hive2://hiveserver:10000/")
+ .option("hoodie.datasource.hive_sync.mode", "jdbc")
+ .option("hoodie.datasource.hive_sync.partition_fields", "dt")
+ .option("hoodie.datasource.hive_sync.partition_extractor_class",
+ "org.apache.hudi.hive.MultiPartKeysValueExtractor")
+ .option("hoodie.datasource.hive_sync.username", "hive")
+ .option("hoodie.datasource.hive_sync.password", "hive")
+}
+
+// Seed two rows in dt='2024-01-01' via Overwrite.
+val seed = Seq(
+ Row(1L, "doc1", Seq(0.1f, 0.2f, 0.3f), "2024-01-01"),
+ Row(2L, "doc2", Seq(0.4f, 0.5f, 0.6f), "2024-01-01")
+)
+val seedDf = spark.createDataFrame(spark.sparkContext.parallelize(seed),
schema)
+applyWriteOpts(seedDf.write).mode(SaveMode.Overwrite).save("/user/hive/warehouse/vector_test_df")
+spark.sql("select id, name, embedding, dt from vector_test_df order by
id").show(10, false)
+println("VECTOR_DF_INSERT_SUCCESS")
+
+// Append-upsert: id=2 mutated, id=3 new in dt='2024-01-02'. New preCombine
+// value (doc2-updated) sorts strictly after the seed (doc2).
+val upsertRows = Seq(
+ Row(2L, "doc2-updated", Seq(0.41f, 0.51f, 0.61f), "2024-01-01"),
+ Row(3L, "doc3", Seq(0.7f, 0.8f, 0.9f), "2024-01-02")
+)
+val upsertDf =
spark.createDataFrame(spark.sparkContext.parallelize(upsertRows), schema)
+applyWriteOpts(upsertDf.write)
+ .option("hoodie.datasource.write.operation", "upsert")
+ .mode(SaveMode.Append)
+ .save("/user/hive/warehouse/vector_test_df")
+spark.sql("select id, name, embedding, dt from vector_test_df order by
id").show(10, false)
+println("VECTOR_DF_UPSERT_SUCCESS")
+
+// Delete id=3 via operation=delete. The DF still needs a schema-compatible
+// VECTOR payload (Hudi uses only the record key/partition for delete).
+val deleteRows = Seq(
+ Row(3L, "doc3", Seq(0.0f, 0.0f, 0.0f), "2024-01-02")
+)
+val deleteDf =
spark.createDataFrame(spark.sparkContext.parallelize(deleteRows), schema)
+applyWriteOpts(deleteDf.write)
+ .option("hoodie.datasource.write.operation", "delete")
+ .mode(SaveMode.Append)
+ .save("/user/hive/warehouse/vector_test_df")
+spark.sql("select id, name, embedding, dt from vector_test_df order by
id").show(10, false)
+println("VECTOR_DF_DELETE_SUCCESS")
+
+println("VECTOR_DF_TEST_SUCCESS")
diff --git a/docker/demo/sparksql-vector-type-sql.commands
b/docker/demo/sparksql-vector-type-sql.commands
new file mode 100644
index 000000000000..e87736a4be2e
--- /dev/null
+++ b/docker/demo/sparksql-vector-type-sql.commands
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// E2E test: VECTOR type with Hive sync via SQL CREATE TABLE path.
+// Runs the full INSERT -> UPDATE -> MERGE -> DELETE lifecycle to exercise
+// the custom-type metadata re-attach path on every write command.
+spark.sql("""
+ CREATE TABLE vector_test (
+ id LONG,
+ name STRING,
+ embedding VECTOR(3),
+ dt STRING
+ ) USING hudi
+ PARTITIONED BY (dt)
+ LOCATION '/user/hive/warehouse/vector_test'
+ TBLPROPERTIES (
+ 'primaryKey' = 'id',
+ 'preCombineField' = 'name',
+ 'hoodie.datasource.hive_sync.enable' = 'true',
+ 'hoodie.datasource.hive_sync.database' = 'default',
+ 'hoodie.datasource.hive_sync.table' = 'vector_test',
+ 'hoodie.datasource.hive_sync.jdbcurl' = 'jdbc:hive2://hiveserver:10000/',
+ 'hoodie.datasource.hive_sync.mode' = 'jdbc',
+ 'hoodie.datasource.hive_sync.partition_fields' = 'dt',
+ 'hoodie.datasource.hive_sync.partition_extractor_class' =
'org.apache.hudi.hive.MultiPartKeysValueExtractor',
+ 'hoodie.datasource.hive_sync.username' = 'hive',
+ 'hoodie.datasource.hive_sync.password' = 'hive'
+ )
+""")
+
+spark.sql("""
+ INSERT INTO vector_test VALUES
+ (1, 'doc1', array(cast(0.1 as float), cast(0.2 as float), cast(0.3 as
float)), '2024-01-01'),
+ (2, 'doc2', array(cast(0.4 as float), cast(0.5 as float), cast(0.6 as
float)), '2024-01-01')
+""")
+spark.sql("select id, name, embedding, dt from vector_test order by
id").show(10, false)
+println("VECTOR_SQL_INSERT_SUCCESS")
+
+// UPDATE goes through castIfNeeded on the VECTOR column; without the reattach
+// path it would fail schema compat with MISSING_UNION_BRANCH.
+spark.sql("""
+ UPDATE vector_test
+ SET embedding = array(cast(0.9 as float), cast(0.8 as float), cast(0.7 as
float)),
+ name = 'doc1-updated'
+ WHERE id = 1
+""")
+spark.sql("select id, name, embedding from vector_test where id = 1").show(10,
false)
+println("VECTOR_SQL_UPDATE_SUCCESS")
+
+// MERGE exercises both MATCHED (UPDATE) and NOT MATCHED (INSERT into new
+// partition dt='2024-01-02'). The USING CTE does not carry the hudi_type
+// marker, so this specifically exercises the reattach path for MERGE.
+spark.sql("""
+ MERGE INTO vector_test t
+ USING (
+ SELECT 2L AS id, 'doc2-merged' AS name,
+ array(cast(0.41 as float), cast(0.51 as float), cast(0.61 as
float)) AS embedding,
+ '2024-01-01' AS dt
+ UNION ALL
+ SELECT 3L AS id, 'doc3' AS name,
+ array(cast(0.7 as float), cast(0.8 as float), cast(0.9 as float))
AS embedding,
+ '2024-01-02' AS dt
+ ) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET t.name = s.name, t.embedding = s.embedding
+ WHEN NOT MATCHED THEN INSERT (id, name, embedding, dt) VALUES (s.id, s.name,
s.embedding, s.dt)
+""")
+spark.sql("select id, name, embedding, dt from vector_test order by
id").show(10, false)
+println("VECTOR_SQL_MERGE_SUCCESS")
+
+spark.sql("DELETE FROM vector_test WHERE id = 3")
+spark.sql("select id, name, dt from vector_test order by id").show(10, false)
+println("VECTOR_SQL_DELETE_SUCCESS")
+
+println("VECTOR_SQL_TEST_SUCCESS")
diff --git a/docker/hoodie/hadoop/hive_base/entrypoint.sh
b/docker/hoodie/hadoop/hive_base/entrypoint.sh
index a3df5e6cf4d7..f048114ff45c 100644
--- a/docker/hoodie/hadoop/hive_base/entrypoint.sh
+++ b/docker/hoodie/hadoop/hive_base/entrypoint.sh
@@ -131,4 +131,4 @@ do
wait_for_it ${i}
done
-exec $@
+exec "$@"
diff --git a/docker/hoodie/hadoop/sparkmaster/master.sh
b/docker/hoodie/hadoop/sparkmaster/master.sh
index 9409cbc0c12e..e437f68ddc00 100644
--- a/docker/hoodie/hadoop/sparkmaster/master.sh
+++ b/docker/hoodie/hadoop/sparkmaster/master.sh
@@ -29,4 +29,4 @@ export SPARK_HOME=/opt/spark
ln -sf /dev/stdout $SPARK_MASTER_LOG/spark-master.out
cd /opt/spark/bin && /opt/spark/sbin/../bin/spark-class
org.apache.spark.deploy.master.Master \
- --ip $SPARK_MASTER_HOST --port $SPARK_MASTER_PORT --webui-port
$SPARK_MASTER_WEBUI_PORT >> $SPARK_MASTER_LOG/spark-master.out
+ --host $SPARK_MASTER_HOST --port $SPARK_MASTER_PORT --webui-port
$SPARK_MASTER_WEBUI_PORT >> $SPARK_MASTER_LOG/spark-master.out
diff --git a/hudi-integ-test/pom.xml b/hudi-integ-test/pom.xml
index bad23d7c38b0..abd44dba109d 100644
--- a/hudi-integ-test/pom.xml
+++ b/hudi-integ-test/pom.xml
@@ -42,12 +42,13 @@
<artifactId>jersey-container-servlet-core</artifactId>
</dependency>
+ <!-- TODO: remove this after we migrate integration tests to
testcontainers (requires docker-java to be >= 3.3.6) -->
<!-- Docker-Java - Test-->
<!-- needs to be at top to force javax.ws version -->
<dependency>
<groupId>com.github.docker-java</groupId>
<artifactId>docker-java</artifactId>
- <version>3.1.2</version>
+ <version>3.3.6</version>
<scope>test</scope>
<exclusions>
<exclusion>
@@ -391,6 +392,19 @@
<artifactId>trino-jdbc</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <!-- not managed in the root pom's dependencyManagement, unlike
testcontainers itself -->
+ <version>${testcontainers.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<properties>
@@ -518,6 +532,38 @@
</build>
<profiles>
+ <profile>
+ <id>integration-tests</id>
+ <!--
+ Tests under org.apache.hudi.integ2.** are testcontainers-based and
manage
+ their own docker stack. The legacy `integration-tests` CI stage brings
up
+ its own docker-compose demo in the same job, which collides on host
ports
+ (e.g. zookeeper 2181). Exclude the integ2 package here so only the
+ dedicated `integration-tests-hive-sync` stage runs them (via
-Dit.test=).
+ -->
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <configuration combine.self="override">
+ <skip>${skipITs}</skip>
+ <includes>
+ <include>**/IT*.java</include>
+ </includes>
+ <excludes>
+ <exclude>**/integ2/**</exclude>
+ </excludes>
+ <systemPropertyVariables>
+
<dynamodb-local.endpoint>${dynamodb-local.endpoint}</dynamodb-local.endpoint>
+
<log4j.configurationFile>${surefire-log4j.file}</log4j.configurationFile>
+ </systemPropertyVariables>
+ <useSystemClassLoader>false</useSystemClassLoader>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
<profile>
<id>m1-mac</id>
<properties>
diff --git
a/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/ContainerProvider.java
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/ContainerProvider.java
new file mode 100644
index 000000000000..0283f24e6b67
--- /dev/null
+++
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/ContainerProvider.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.integ2.testcontainers;
+
+import org.testcontainers.containers.ContainerState;
+
+/**
+ * Interface for providing access to containers by service name.
+ * This abstraction allows different implementations (e.g., ComposeContainer,
individual containers).
+ */
+public interface ContainerProvider {
+
+ /**
+ * Get a container by its service name.
+ *
+ * @param serviceName the full compose service instance name as
Testcontainers resolves it (e.g. "adhoc-1-1")
+ * @return the container state
+ * @throws IllegalStateException if container not found
+ */
+ ContainerState getContainer(String serviceName);
+}
diff --git
a/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/ITTestBaseTestcontainers.java
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/ITTestBaseTestcontainers.java
new file mode 100644
index 000000000000..51fdd57b4bfb
--- /dev/null
+++
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/ITTestBaseTestcontainers.java
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.integ2.testcontainers;
+
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.integ2.testcontainers.service.HiveService;
+import org.apache.hudi.integ2.testcontainers.service.SparkService;
+
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeAll;
+import org.testcontainers.containers.ComposeContainer;
+import org.testcontainers.containers.ContainerState;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+
+import static
org.apache.hudi.integ2.testcontainers.TestcontainersConfig.Containers;
+import static
org.apache.hudi.integ2.testcontainers.TestcontainersConfig.Network;
+import static
org.apache.hudi.integ2.testcontainers.TestcontainersConfig.SystemProps;
+import static
org.apache.hudi.integ2.testcontainers.TestcontainersConfig.Timeouts;
+
+/**
+ * Base test class for integration tests using Testcontainers with Docker
Compose.
+ * Uses a "Service" pattern, where each external service (Hive, Spark, etc.) is
+ * represented by a dedicated service object. This provides better separation
of
+ * concerns and a cleaner API for tests.
+ */
+@Slf4j
+@Testcontainers
+public abstract class ITTestBaseTestcontainers implements ContainerProvider {
+
+ protected static ComposeContainer environment;
+
+ // Service objects for interacting with different components
+ protected HiveService hive;
+ protected SparkService sparkAdhoc1;
+
+ @BeforeAll
+ public static void setupDockerCompose() {
+ String composeFilePath =
createProcessedComposeFilePath(getDockerComposeFilePath());
+ String hudiWorkspace = getHudiWorkspace();
+
+ log.info("Starting Docker Compose environment");
+ log.info("Compose file: {}", composeFilePath);
+ log.info("HUDI_WS: {}", hudiWorkspace);
+
+ environment = new ComposeContainer(new File(composeFilePath))
+ .withEnv("HUDI_WS", hudiWorkspace)
+ .withExposedService(Containers.SPARK_MASTER,
Network.SPARK_MASTER_WEB_UI_PORT,
+ Wait.forListeningPort().forPorts(Network.SPARK_MASTER_WEB_UI_PORT)
+ .withStartupTimeout(Timeouts.CONTAINER_STARTUP))
+ .withStartupTimeout(Timeouts.CONTAINER_STARTUP);
+ environment.start();
+
+ log.info("Docker Compose environment started successfully");
+ log.info("All containers verified and running");
+ }
+
+ /**
+ * Tear down the compose stack between test classes. The docker-compose
files publish
+ * host ports directly (zookeeper 2181, spark 7077, …), so leaving one
+ * stack up would make the next class's `@BeforeAll` collide on those host
ports.
+ * Testcontainers' Ryuk reaper only fires at JVM shutdown, which is too late
when
+ * failsafe reuses a JVM across classes.
+ */
+ @AfterAll
+ public static void tearDownDockerCompose() {
+ if (environment != null) {
+ log.info("Stopping Docker Compose environment");
+ try {
+ environment.stop();
+ } finally {
+ environment = null;
+ }
+ }
+ }
+
+ /**
+ * Initialize service objects. Should be called in @BeforeEach or
constructor of test class.
+ */
+ protected void initializeServices() {
+ this.hive = new HiveService(this);
+ this.sparkAdhoc1 = new SparkService(this, Containers.ADHOC_1);
+ }
+
+ /**
+ * Skips the test unless the active docker-compose prefix points to a Spark
4.x stack.
+ * Use for tests that rely on Spark 4.0+ only features (e.g. VARIANT type).
+ */
+ protected static void assumeSpark4Compose() {
+ Assumptions.assumeTrue(isSpark4Compose(),
+ "Test requires a Spark 4.x compose stack; active prefix is '"
+ + System.getProperty(SystemProps.COMPOSE_PREFIX,
SystemProps.DEFAULT_COMPOSE_PREFIX) + "'");
+ }
+
+ /**
+ * Non-assumption variant of {@link #assumeSpark4Compose()}: returns {@code
true} when the
+ * active compose prefix points at a Spark 4.x stack. Use in {@code
@BeforeAll} seeding
+ * to conditionally run Spark 4-only fixtures (e.g. VARIANT) without
aborting the whole
+ * test class on a Spark 3.5 run.
+ */
+ protected static boolean isSpark4Compose() {
+ String composePrefix = System.getProperty(SystemProps.COMPOSE_PREFIX,
SystemProps.DEFAULT_COMPOSE_PREFIX);
+ return composePrefix.contains(SystemProps.SPARK_4_PREFIX_TOKEN);
+ }
+
+ /**
+ * Waits for HDFS namenode to be ready by retrying the safemode wait command.
+ * The namenode may take some time to start after Docker Compose reports
containers as running.
+ */
+ protected void waitForHdfs() throws Exception {
+ for (int i = 1; i <= Timeouts.HDFS_MAX_RETRIES; i++) {
+ try {
+ sparkAdhoc1.executeShellCommand("hdfs dfsadmin -safemode
wait").expectToSucceed();
+ log.info("HDFS namenode is ready");
+ return;
+ } catch (Throwable e) {
+ if (i == Timeouts.HDFS_MAX_RETRIES) {
+ throw new RuntimeException(
+ "HDFS namenode did not become ready after " +
Timeouts.HDFS_MAX_RETRIES + " retries", e);
+ }
+ log.info("Waiting for HDFS namenode to be ready (attempt {}/{})", i,
Timeouts.HDFS_MAX_RETRIES);
+ Thread.sleep(Timeouts.HDFS_RETRY_INTERVAL.toMillis());
+ }
+ }
+ }
+
+ /**
+ * Get a container by service name from the Docker Compose environment.
+ * Docker Compose appends _1 suffix to service names.
+ * Implements ContainerProvider interface.
+ */
+ @Override
+ public ContainerState getContainer(String serviceName) {
+ try {
+ return environment.getContainerByServiceName(serviceName)
+ .orElseThrow(() -> new IllegalStateException("Container not found: "
+ serviceName));
+ } catch (IllegalStateException e) {
+ log.error("Failed to get container: {}", serviceName, e);
+ throw e;
+ }
+ }
+
+ private static String getHudiWorkspace() {
+ String projectDir = System.getProperty("user.dir");
+ return new File(projectDir, "..").getAbsolutePath();
+ }
+
+ private static String getDockerComposeFilePath() {
+ String projectDir = System.getProperty("user.dir");
+ String os = System.getProperty("os.name").toLowerCase();
+ String arch = System.getProperty("os.arch").toLowerCase();
+ String composePrefix = System.getProperty(SystemProps.COMPOSE_PREFIX,
SystemProps.DEFAULT_COMPOSE_PREFIX);
+
+ // Determine which compose file to use based on OS and architecture
+ boolean isMacArm64 = os.contains("mac") && arch.contains("aarch64");
+ String archSuffix = isMacArm64 ? "_arm64" : "_amd64";
+ File dockerComposeFile = new File(projectDir,
+ TestcontainersConfig.Paths.COMPOSE_DIR + composePrefix + archSuffix +
".yml");
+ if (!dockerComposeFile.isFile() || !dockerComposeFile.exists()) {
+ throw new HoodieException(String.format("%s does not exist",
dockerComposeFile.getAbsolutePath()));
+ }
+ return dockerComposeFile.getAbsolutePath();
+ }
+
+ private static String getHadoopEnvFilePath() {
+ return new File(System.getProperty("user.dir"),
"../docker/compose/hadoop.env").getAbsolutePath();
+ }
+
+ /**
+ * Reads the original docker-compose file, removes all 'container_name'
directives, and returns a temporary file containing the modified content.
Including Testcontainers in the docker-compose file
+ * will cause ContainerLaunchExceptions to be thrown.
+ * <p>
+ * Any env_file will normalize/canonicalize to the temporary directory used.
This function will make a copy of hadoop.env into the temporary directory to
ensure that no error is thrown.
+ */
+ private static String createProcessedComposeFilePath(String composeFile) {
+ try {
+ // Read all bytes from the file and convert to a String
+ byte[] bytes = Files.readAllBytes(Paths.get(composeFile));
+ String originalContent = new String(bytes, StandardCharsets.UTF_8);
+
+ // Use a regular expression to find and remove all lines containing
'container_name'
+ String modifiedContent =
originalContent.replaceAll("(?m)^\\s*container_name:.*$", "");
+
+ // Create a temporary file to hold our modified configuration
+ Path tempDir = Files.createTempDirectory("hudi-test-compose-");
+
+ // Delete the temp dir at JVM exit. File.delete() cannot remove a
non-empty directory,
+ // and deleteOnExit runs LIFO, so the files below are registered too:
they are deleted
+ // first, leaving the dir empty for its own delete.
+ tempDir.toFile().deleteOnExit();
+
+ // Write the modified content to the temporary file as a byte array.
+ File tempDockerComposeFile = new File(tempDir.toFile(),
"docker-compose.yml");
+ Files.write(tempDockerComposeFile.toPath(),
modifiedContent.getBytes(StandardCharsets.UTF_8));
+ tempDockerComposeFile.deleteOnExit();
+
+ // tempDir is used as the working directory, docker-compose will look
for hadoop.env in the SAME temp directory
+ // Copy hadoop.env into the SAME temp directory
+ Path destHadoopEnvPath = tempDir.resolve("hadoop.env");
+ Path originalHadoopEnvPath =
Paths.get(getHadoopEnvFilePath()).toAbsolutePath().normalize();
+ Files.copy(originalHadoopEnvPath, destHadoopEnvPath,
StandardCopyOption.REPLACE_EXISTING);
+ destHadoopEnvPath.toFile().deleteOnExit();
+
+ // Return the temporary file for Testcontainers to use
+ return tempDockerComposeFile.toPath().toAbsolutePath().toString();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to process the docker-compose file",
e);
+ }
+ }
+}
diff --git
a/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/ITTestCustomTypeHiveSync.java
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/ITTestCustomTypeHiveSync.java
new file mode 100644
index 000000000000..3b3cbc5e5e0a
--- /dev/null
+++
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/ITTestCustomTypeHiveSync.java
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.integ2.testcontainers;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+import static org.apache.hudi.integ2.testcontainers.TestcontainersConfig.Paths;
+
+/**
+ * End-to-end Hive sync coverage for Hudi's custom logical types (VECTOR,
BLOB) and the
+ * Spark 4.0 VARIANT type, running against a real Hive metastore via the
Testcontainers
+ * harness.
+ *
+ * Each type is exercised through both paths:
+ * - SQL CREATE TABLE (`*-sql.commands`) - table name `<type>_test`
+ * - DataFrame writer API (`*-df.commands`) - table name `<type>_test_df`
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class ITTestCustomTypeHiveSync extends ITTestBaseTestcontainers {
+
+ // BLOB
+ private static final String BLOB_SQL_TEST_BASE_PATH =
"/user/hive/warehouse/blob_test";
+ private static final String BLOB_DF_TEST_BASE_PATH =
"/user/hive/warehouse/blob_test_df";
+ private static final String SPARKSQL_BLOB_TYPE_SQL_COMMANDS = Paths.DEMO_DIR
+ "/sparksql-blob-type-sql.commands";
+ private static final String SPARKSQL_BLOB_TYPE_DF_COMMANDS = Paths.DEMO_DIR
+ "/sparksql-blob-type-df.commands";
+
+ // VARIANT
+ private static final String VARIANT_SQL_TEST_BASE_PATH =
"/user/hive/warehouse/variant_test";
+ private static final String VARIANT_DF_TEST_BASE_PATH =
"/user/hive/warehouse/variant_test_df";
+ private static final String SPARKSQL_VARIANT_TYPE_SQL_COMMANDS =
Paths.DEMO_DIR + "/sparksql-variant-type-sql.commands";
+ private static final String SPARKSQL_VARIANT_TYPE_DF_COMMANDS =
Paths.DEMO_DIR + "/sparksql-variant-type-df.commands";
+
+ // VECTOR
+ private static final String VECTOR_SQL_TEST_BASE_PATH =
"/user/hive/warehouse/vector_test";
+ private static final String VECTOR_DF_TEST_BASE_PATH =
"/user/hive/warehouse/vector_test_df";
+ private static final String SPARKSQL_VECTOR_TYPE_SQL_COMMANDS =
Paths.DEMO_DIR + "/sparksql-vector-type-sql.commands";
+ private static final String SPARKSQL_VECTOR_TYPE_DF_COMMANDS =
Paths.DEMO_DIR + "/sparksql-vector-type-df.commands";
+
+ // All test paths cleaned in a single `hdfs dfs -rm -R -f` call; -f silently
skips
+ // non-existent paths so tests that don't create every table don't fail
teardown.
+ private static final String CLEANUP_PATHS_JOINED = String.join(" ",
+ BLOB_SQL_TEST_BASE_PATH, BLOB_DF_TEST_BASE_PATH,
+ VARIANT_SQL_TEST_BASE_PATH, VARIANT_DF_TEST_BASE_PATH,
+ VECTOR_SQL_TEST_BASE_PATH, VECTOR_DF_TEST_BASE_PATH);
+
+ /**
+ * Run idempotent demo setup once per test class instead of once per @Test.
The script
+ * (mkdir -p, cp, copyFromLocal -f, chmod +x) is safe to run a single time
and saves
+ * one shell exec round-trip per test. Requires @TestInstance(PER_CLASS) so
this method
+ * can be a non-static instance method that uses sparkAdhoc1.
+ */
+ @BeforeAll
+ public void setupOnce() throws Exception {
+ initializeServices();
+ waitForHdfs();
+ sparkAdhoc1.executeShellCommand("/bin/bash " +
Paths.DEMO_SETUP).expectToSucceed();
+ }
+
+ @AfterEach
+ public void clean() throws Exception {
+ sparkAdhoc1.executeShellCommand("hdfs dfs -rm -R -f " +
CLEANUP_PATHS_JOINED)
+ .expectToSucceed();
+ }
+
+ // ---------- BLOB ----------
+
+ @Test
+ public void testBlobTypeWithHiveSyncSQL() throws Exception {
+ sparkAdhoc1.executeSQLFile(SPARKSQL_BLOB_TYPE_SQL_COMMANDS)
+ .expectToSucceed()
+ .assertStdOutContainsLine("BLOB_SQL_INSERT_SUCCESS")
+ .assertStdOutContainsLine("BLOB_SQL_UPDATE_SUCCESS")
+ .assertStdOutContainsLine("BLOB_SQL_MERGE_SUCCESS")
+ .assertStdOutContainsLine("BLOB_SQL_DELETE_SUCCESS")
+ .assertStdOutContainsLine("BLOB_SQL_TEST_SUCCESS");
+
+ hive.execute("DESCRIBE default.blob_test")
+ .expectToSucceed()
+ .assertStdOutContains("blob_data");
+
+ // MERGE added dt=2024-01-02; DELETE removed the row but kept the
partition metadata.
+ hive.execute("SHOW PARTITIONS default.blob_test")
+ .expectToSucceed()
+ .assertStdOutContains("dt=2024-01-01")
+ .assertStdOutContains("dt=2024-01-02");
+
+ // Post-DELETE final row count is 2 (id=1 updated, id=2 merged; id=3
deleted).
+ hive.execute("SELECT concat('HIVE_COUNT=', count(*)) FROM
default.blob_test")
+ .expectToSucceed()
+ .assertStdOutContains("HIVE_COUNT=2");
+
+ // Project a nested struct field through the Hive serde to verify the BLOB
+ // struct layout round-trips correctly (count() and DESCRIBE do not).
+ hive.execute(
+ "SELECT concat('BLOB_TYPE=', blob_data.type) FROM default.blob_test
WHERE id = 1")
+ .expectToSucceed()
+ .assertStdOutContains("BLOB_TYPE=OUT_OF_LINE");
+ }
+
+ @Test
+ public void testBlobTypeWithHiveSyncDataFrameAPI() throws Exception {
+ sparkAdhoc1.executeSQLFile(SPARKSQL_BLOB_TYPE_DF_COMMANDS)
+ .expectToSucceed()
+ .assertStdOutContainsLine("BLOB_DF_INSERT_SUCCESS")
+ .assertStdOutContainsLine("BLOB_DF_UPSERT_SUCCESS")
+ .assertStdOutContainsLine("BLOB_DF_DELETE_SUCCESS")
+ .assertStdOutContainsLine("BLOB_DF_TEST_SUCCESS");
+
+ hive.execute("DESCRIBE default.blob_test_df")
+ .expectToSucceed()
+ .assertStdOutContains("blob_data");
+
+ hive.execute("SHOW PARTITIONS default.blob_test_df")
+ .expectToSucceed()
+ .assertStdOutContains("dt=2024-01-01")
+ .assertStdOutContains("dt=2024-01-02");
+
+ hive.execute("SELECT concat('HIVE_COUNT=', count(*)) FROM
default.blob_test_df")
+ .expectToSucceed()
+ .assertStdOutContains("HIVE_COUNT=2");
+
+ // DF path seed used the INLINE struct branch and id=1 is never mutated,
+ // so projecting blob_data.type through the Hive serde should return
INLINE.
+ hive.execute(
+ "SELECT concat('BLOB_TYPE=', blob_data.type) FROM default.blob_test_df
WHERE id = 1")
+ .expectToSucceed()
+ .assertStdOutContains("BLOB_TYPE=INLINE");
+ }
+
+ // ---------- VARIANT (Spark 4.x only) ----------
+
+ @Test
+ public void testVariantTypeWithHiveSyncSQL() throws Exception {
+ assumeSpark4Compose();
+ sparkAdhoc1.executeSQLFile(SPARKSQL_VARIANT_TYPE_SQL_COMMANDS)
+ .expectToSucceed()
+ .assertStdOutContainsLine("VARIANT_SQL_INSERT_SUCCESS")
+ .assertStdOutContainsLine("VARIANT_SQL_UPDATE_SUCCESS")
+ .assertStdOutContainsLine("VARIANT_SQL_MERGE_SUCCESS")
+ .assertStdOutContainsLine("VARIANT_SQL_DELETE_SUCCESS")
+ .assertStdOutContainsLine("VARIANT_SQL_TEST_SUCCESS");
+
+ hive.execute("DESCRIBE default.variant_test")
+ .expectToSucceed()
+ .assertStdOutContains("variant_data");
+
+ hive.execute("SHOW PARTITIONS default.variant_test")
+ .expectToSucceed()
+ .assertStdOutContains("dt=2024-01-01")
+ .assertStdOutContains("dt=2024-01-02");
+
+ // count(*) does not deserialize the variant column, so it is safe even if
+ // the Hive serde can't project the variant payload.
+ hive.execute("SELECT concat('HIVE_COUNT=', count(*)) FROM
default.variant_test")
+ .expectToSucceed()
+ .assertStdOutContains("HIVE_COUNT=2");
+ }
+
+ @Test
+ public void testVariantTypeWithHiveSyncDataFrameAPI() throws Exception {
+ assumeSpark4Compose();
+ sparkAdhoc1.executeSQLFile(SPARKSQL_VARIANT_TYPE_DF_COMMANDS)
+ .expectToSucceed()
+ .assertStdOutContainsLine("VARIANT_DF_INSERT_SUCCESS")
+ .assertStdOutContainsLine("VARIANT_DF_UPSERT_SUCCESS")
+ .assertStdOutContainsLine("VARIANT_DF_DELETE_SUCCESS")
+ .assertStdOutContainsLine("VARIANT_DF_TEST_SUCCESS");
+
+ hive.execute("DESCRIBE default.variant_test_df")
+ .expectToSucceed()
+ .assertStdOutContains("variant_data");
+
+ hive.execute("SHOW PARTITIONS default.variant_test_df")
+ .expectToSucceed()
+ .assertStdOutContains("dt=2024-01-01")
+ .assertStdOutContains("dt=2024-01-02");
+
+ hive.execute("SELECT concat('HIVE_COUNT=', count(*)) FROM
default.variant_test_df")
+ .expectToSucceed()
+ .assertStdOutContains("HIVE_COUNT=2");
+ }
+
+ // ---------- VECTOR ----------
+
+ @Test
+ public void testVectorTypeWithHiveSyncSQL() throws Exception {
+ sparkAdhoc1.executeSQLFile(SPARKSQL_VECTOR_TYPE_SQL_COMMANDS)
+ .expectToSucceed()
+ .assertStdOutContainsLine("VECTOR_SQL_INSERT_SUCCESS")
+ .assertStdOutContainsLine("VECTOR_SQL_UPDATE_SUCCESS")
+ .assertStdOutContainsLine("VECTOR_SQL_MERGE_SUCCESS")
+ .assertStdOutContainsLine("VECTOR_SQL_DELETE_SUCCESS")
+ .assertStdOutContainsLine("VECTOR_SQL_TEST_SUCCESS");
+
+ hive.execute("DESCRIBE default.vector_test")
+ .expectToSucceed()
+ .assertStdOutContains("embedding")
+ .assertStdOutContains("binary");
+
+ hive.execute("SHOW PARTITIONS default.vector_test")
+ .expectToSucceed()
+ .assertStdOutContains("dt=2024-01-01")
+ .assertStdOutContains("dt=2024-01-02");
+
+ hive.execute("SELECT concat('HIVE_COUNT=', count(*)) FROM
default.vector_test")
+ .expectToSucceed()
+ .assertStdOutContains("HIVE_COUNT=2");
+
+ // VECTOR(3) is stored on disk as fixed_len_byte_array(12) and mapped to
+ // Hive BINARY (per RFC-99). length() on the binary column confirms the
+ // bytes round-trip through the Hive serde at the correct width: 3 floats
+ // x 4 bytes each = 12. Any flip to array<float> would return 3 instead.
+ hive.execute(
+ "SELECT concat('VEC_LEN=', length(embedding)) FROM default.vector_test
WHERE id = 1")
+ .expectToSucceed()
+ .assertStdOutContains("VEC_LEN=12");
+ }
+
+ @Test
+ public void testVectorTypeWithHiveSyncDataFrameAPI() throws Exception {
+ sparkAdhoc1.executeSQLFile(SPARKSQL_VECTOR_TYPE_DF_COMMANDS)
+ .expectToSucceed()
+ .assertStdOutContainsLine("VECTOR_DF_INSERT_SUCCESS")
+ .assertStdOutContainsLine("VECTOR_DF_UPSERT_SUCCESS")
+ .assertStdOutContainsLine("VECTOR_DF_DELETE_SUCCESS")
+ .assertStdOutContainsLine("VECTOR_DF_TEST_SUCCESS");
+
+ hive.execute("DESCRIBE default.vector_test_df")
+ .expectToSucceed()
+ .assertStdOutContains("embedding")
+ .assertStdOutContains("binary");
+
+ hive.execute("SHOW PARTITIONS default.vector_test_df")
+ .expectToSucceed()
+ .assertStdOutContains("dt=2024-01-01")
+ .assertStdOutContains("dt=2024-01-02");
+
+ hive.execute("SELECT concat('HIVE_COUNT=', count(*)) FROM
default.vector_test_df")
+ .expectToSucceed()
+ .assertStdOutContains("HIVE_COUNT=2");
+
+ hive.execute(
+ "SELECT concat('VEC_LEN=', length(embedding)) FROM
default.vector_test_df WHERE id = 1")
+ .expectToSucceed()
+ .assertStdOutContains("VEC_LEN=12");
+ }
+}
diff --git
a/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/TestcontainersConfig.java
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/TestcontainersConfig.java
new file mode 100644
index 000000000000..ad403ff6cccd
--- /dev/null
+++
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/TestcontainersConfig.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.integ2.testcontainers;
+
+import java.time.Duration;
+import java.util.List;
+
+/**
+ * Central configuration for the integ2 Testcontainers harness. Every constant
that
+ * describes "how this harness expects the compose environment to look"
belongs here:
+ * container service names, container-side paths, network endpoints, timeouts,
and
+ * the keys / defaults of system properties that tune the harness.
+ *
+ * <p>Per-test fixture data (table paths, per-test {@code .commands} scripts)
does
+ * not belong here, those stay in the relevant test class, though they should
build
+ * their common prefix from {@link Paths#DEMO_DIR} / {@link Paths#WS_ROOT}
rather
+ * than inlining the path literal.
+ */
+public final class TestcontainersConfig {
+
+ private TestcontainersConfig() {
+ }
+
+ /** Docker Compose service names. Must match the compose YAML verbatim. */
+ public static final class Containers {
+ public static final String HIVESERVER = "hiveserver";
+ public static final String SPARK_MASTER = "sparkmaster";
+ // Testcontainers appends the replica index, so the adhoc services resolve
as "<name>-1".
+ public static final String ADHOC_1 = "adhoc-1-1";
+
+ private Containers() {
+ }
+ }
+
+ /** Paths inside the containers (absolute) and the host-side compose
directory. */
+ public static final class Paths {
+ public static final String WS_ROOT = "/var/hoodie/ws";
+ public static final String DEMO_DIR = WS_ROOT + "/docker/demo";
+ public static final String DEMO_SETUP = DEMO_DIR +
"/setup_demo_container.sh";
+ public static final String HIVE_TARGET = WS_ROOT +
"/docker/hoodie/hadoop/hive_base/target";
+ public static final String SPARK_BUNDLE = HIVE_TARGET +
"/hoodie-spark-bundle.jar";
+ public static final String HADOOP_CONF_DIR = "/etc/hadoop";
+ /** Host-side, relative to the hudi-integ-test module working directory. */
+ public static final String COMPOSE_DIR = "../docker/compose/";
+
+ private Paths() {
+ }
+ }
+
+ /** Network endpoints the harness exposes to tests. */
+ public static final class Network {
+ public static final int SPARK_MASTER_WEB_UI_PORT = 8080;
+
+ private Network() {
+ }
+ }
+
+ /** Waits and timeouts used by the harness. */
+ public static final class Timeouts {
+ public static final Duration CONTAINER_STARTUP = Duration.ofMinutes(5);
+ public static final int HDFS_MAX_RETRIES = 12;
+ public static final Duration HDFS_RETRY_INTERVAL = Duration.ofSeconds(10);
+
+ private Timeouts() {
+ }
+ }
+
+ /** System-property keys and their defaults (read via {@link
System#getProperty}). */
+ public static final class SystemProps {
+ public static final String COMPOSE_PREFIX = "spark.docker.compose.prefix";
+ public static final String DEFAULT_COMPOSE_PREFIX =
"docker-compose_hadoop340_hive2310_spark402";
+ /** Substring present in compose prefixes that run Spark 4.x (e.g.
"...spark402"). */
+ public static final String SPARK_4_PREFIX_TOKEN = "spark4";
+
+ /**
+ * Flip to {@code true} (e.g. {@code -Dhudi.integ.hive.verbose=true}) to
route
+ * Hive logs to the console so exception stack traces show up in test
output.
+ */
+ public static final String HIVE_VERBOSE = "hudi.integ.hive.verbose";
+
+ /**
+ * Hiveconf entries applied only when {@link #HIVE_VERBOSE} is enabled.
+ * See {@code HiveService#execute} for the per-flag rationale.
+ */
+ public static final List<String> VERBOSE_HIVECONFS =
+ List.of("hive.root.logger=INFO,console",
+ "hive.exec.mode.local.auto=false",
+ "hive.log.explain.output=true",
+ "hive.server2.logging.operation.verbose=true");
+
+ private SystemProps() {
+ }
+ }
+}
diff --git
a/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/command/CommandExecutor.java
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/command/CommandExecutor.java
new file mode 100644
index 000000000000..f2f01a006c1a
--- /dev/null
+++
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/command/CommandExecutor.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.integ2.testcontainers.command;
+
+import lombok.extern.slf4j.Slf4j;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.ContainerState;
+
+/**
+ * A utility class for executing commands within a given Testcontainer.
+ * This class is stateless regarding the command being executed but is tied to
a specific container.
+ */
+@Slf4j
+public final class CommandExecutor {
+
+ private final ContainerState container;
+
+ public CommandExecutor(ContainerState container) {
+ this.container = container;
+ }
+
+ /**
+ * Execute a command in the executor's container.
+ */
+ public CommandResult executeCommand(String... command) throws Exception {
+ String containerIdentifier = getContainerIdentifier(container);
+ String commandStr = String.join(" ", command);
+
+ log.info("==> [{}] Executing: {}", containerIdentifier, commandStr);
+
+ long startTime = System.currentTimeMillis();
+ Container.ExecResult result = container.execInContainer(command);
+ long duration = System.currentTimeMillis() - startTime;
+
+ int exitCode = result.getExitCode();
+ log.info("<== [{}] Exit code: {} ({}ms)", containerIdentifier, exitCode,
duration);
+
+ if (exitCode != 0) {
+ log.error("STDOUT:\n{}", result.getStdout());
+ log.error("STDERR:\n{}", result.getStderr());
+ } else if (log.isDebugEnabled()) {
+ log.debug("STDOUT:\n{}", result.getStdout());
+ if (!result.getStderr().isEmpty()) {
+ log.debug("STDERR:\n{}", result.getStderr());
+ }
+ }
+
+ return new CommandResult(result);
+ }
+
+ /**
+ * Execute a shell command string.
+ */
+ public CommandResult executeCommandString(String cmd) throws Exception {
+ String[] cmdArray = {"/bin/bash", "-c", cmd};
+ return executeCommand(cmdArray);
+ }
+
+ /**
+ * Get a readable identifier for the container.
+ */
+ private String getContainerIdentifier(ContainerState container) {
+ String containerName = container.getContainerInfo().getName();
+ if (containerName != null && !containerName.isEmpty()) {
+ // Container names start with '/', so remove it
+ String cleanName = containerName.startsWith("/") ?
containerName.substring(1) : containerName;
+ return cleanName + ":" + container.getContainerId().substring(0, 8);
+ }
+ return container.getContainerId().substring(0, 12);
+ }
+}
diff --git
a/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/command/CommandResult.java
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/command/CommandResult.java
new file mode 100644
index 000000000000..8e65e9aebcba
--- /dev/null
+++
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/command/CommandResult.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.integ2.testcontainers.command;
+
+import lombok.AllArgsConstructor;
+import org.testcontainers.containers.Container;
+
+import java.util.regex.Pattern;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * A dedicated class to hold the result of a command execution and provide
+ * fluent assertion methods for cleaner tests.
+ */
+@AllArgsConstructor
+public class CommandResult {
+
+ // Spark 4.0 (Scala 2.13) spark-shell runs under a dumb terminal when stdin
+ // is piped from a file, which flushes the next `scala> ` prompt onto the
+ // same line as preceding async println output (e.g. `scala> MARKER`).
+ // Stripping one-or-more leading `scala>\s+` prefixes normalizes those lines
+ // back to the bare sentinel while leaving Spark 3.5 output (sentinel already
+ // on its own line) unchanged.
+ private static final Pattern REPL_PROMPT_PREFIX =
Pattern.compile("^(scala>\\s+)+");
+
+ private final String stdout;
+ private final String stderr;
+ private final int exitCode;
+
+ public CommandResult(Container.ExecResult execResult) {
+ this.stdout = execResult.getStdout();
+ this.stderr = execResult.getStderr();
+ this.exitCode = execResult.getExitCode();
+ }
+
+ /**
+ * Asserts that the command's exit code is 0 (success).
+ *
+ * @return The same {@link CommandResult} instance for chaining assertions.
+ * @throws AssertionError if the exit code is not 0.
+ */
+ public CommandResult expectToSucceed() {
+ assertEquals(0, exitCode,
+ String.format("Command failed with exit code %d. Stderr: %s",
exitCode, stderr));
+ return this;
+ }
+
+ /**
+ * Asserts that the standard output contains a specific substring exactly
once.
+ *
+ * @param expectedSubstring The substring to search for.
+ * @return The same {@link CommandResult} instance for chaining assertions.
+ */
+ public CommandResult assertStdOutContains(String expectedSubstring) {
+ return assertStdOutContains(expectedSubstring, 1);
+ }
+
+ /**
+ * Asserts that the standard output contains a specific substring an exact
number of times.
+ *
+ * @param expectedSubstring The substring to search for.
+ * @param times The exact number of times the substring is expected to
appear.
+ * @return The same {@link CommandResult} instance for chaining assertions.
+ */
+ public CommandResult assertStdOutContains(String expectedSubstring, int
times) {
+ // Normalize whitespace for more robust matching
+ String stdOutSingleSpaced = stdout.replaceAll("[\\s]+", " ").trim();
+ String expectedOutput = expectedSubstring.replaceAll("[\\s]+", " ").trim();
+
+ int lastIndex = 0;
+ int count = 0;
+ while (lastIndex != -1) {
+ lastIndex = stdOutSingleSpaced.indexOf(expectedOutput, lastIndex);
+ if (lastIndex != -1) {
+ count++;
+ lastIndex += expectedOutput.length();
+ }
+ }
+ assertEquals(times, count,
+ String.format("Expected to find substring '%s' %d times, but found %d.
Full stdout: %s",
+ expectedOutput, times, count, stdout));
+ return this;
+ }
+
+ /**
+ * Asserts that the standard output contains a line exactly equal to the
expected value
+ * (after trimming and after stripping any leading {@code scala> } REPL
prompt prefix),
+ * appearing exactly once. Use this for REPL sentinel markers emitted via
+ * {@code println(...)}. Handles both Scala 2.12's spark-shell echoing the
input line
+ * back (Spark 3.5) and Scala 2.13's dumb-terminal mode prefixing sentinel
output with
+ * {@code scala> } on the same line (Spark 4.0).
+ *
+ * @param expectedLine The exact line content to match (after trim and
prompt strip).
+ * @return The same {@link CommandResult} instance for chaining assertions.
+ */
+ public CommandResult assertStdOutContainsLine(String expectedLine) {
+ return assertStdOutContainsLine(expectedLine, 1);
+ }
+
+ /**
+ * Asserts that the standard output contains a line exactly equal to the
expected value
+ * (after trimming and after stripping any leading {@code scala> } REPL
prompt prefix),
+ * appearing an exact number of times.
+ *
+ * @param expectedLine The exact line content to match (after trim and
prompt strip).
+ * @param times The exact number of matching lines expected.
+ * @return The same {@link CommandResult} instance for chaining assertions.
+ */
+ public CommandResult assertStdOutContainsLine(String expectedLine, int
times) {
+ String expected = expectedLine.trim();
+ long count = stdout.lines()
+ .map(line -> REPL_PROMPT_PREFIX.matcher(line).replaceFirst("").trim())
+ .filter(expected::equals)
+ .count();
+ assertEquals(times, count,
+ String.format("Expected line '%s' %d times, but found %d. Full stdout:
%s",
+ expected, times, count, stdout));
+ return this;
+ }
+}
diff --git
a/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/service/HiveService.java
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/service/HiveService.java
new file mode 100644
index 000000000000..5c5151809a7a
--- /dev/null
+++
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/service/HiveService.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.integ2.testcontainers.service;
+
+import org.apache.hudi.integ2.testcontainers.ContainerProvider;
+import org.apache.hudi.integ2.testcontainers.TestcontainersConfig;
+import org.apache.hudi.integ2.testcontainers.command.CommandExecutor;
+import org.apache.hudi.integ2.testcontainers.command.CommandResult;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A service wrapper for the Hive container.
+ * This class is responsible for all interactions with the Hive service,
+ * including executing commands and managing files.
+ */
+public class HiveService {
+
+ private final CommandExecutor executor;
+ private final boolean verbose;
+
+ public HiveService(ContainerProvider provider) {
+ this(provider,
Boolean.getBoolean(TestcontainersConfig.SystemProps.HIVE_VERBOSE));
+ }
+
+ /**
+ * Visible-for-tests overload so callers can toggle verbose mode without
+ * setting the system property at JVM start time.
+ */
+ public HiveService(ContainerProvider provider, boolean verbose) {
+ this.executor = new
CommandExecutor(provider.getContainer(TestcontainersConfig.Containers.HIVESERVER));
+ this.verbose = verbose;
+ }
+
+ /**
+ * Execute a Hive command and return the result.
+ */
+ public CommandResult execute(String hiveCommand) throws Exception {
+ List<String> hiveCmd = new ArrayList<>();
+ hiveCmd.add("hive");
+ hiveCmd.add("--hiveconf");
+
hiveCmd.add("hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat");
+ hiveCmd.add("--hiveconf");
+ hiveCmd.add("hive.stats.autogather=false");
+ if (verbose) {
+ for (String kv : TestcontainersConfig.SystemProps.VERBOSE_HIVECONFS) {
+ hiveCmd.add("--hiveconf");
+ hiveCmd.add(kv);
+ }
+ }
+ hiveCmd.add("-e");
+ hiveCmd.add(hiveCommand);
+ return executor.executeCommand(hiveCmd.toArray(new String[0]));
+ }
+}
diff --git
a/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/service/SparkService.java
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/service/SparkService.java
new file mode 100644
index 000000000000..1c105402d644
--- /dev/null
+++
b/hudi-integ-test/src/test/java/org/apache/hudi/integ2/testcontainers/service/SparkService.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.integ2.testcontainers.service;
+
+import org.apache.hudi.integ2.testcontainers.ContainerProvider;
+import org.apache.hudi.integ2.testcontainers.TestcontainersConfig;
+import org.apache.hudi.integ2.testcontainers.command.CommandExecutor;
+import org.apache.hudi.integ2.testcontainers.command.CommandResult;
+
+/**
+ * A service wrapper for the Spark container.
+ * This class is responsible for all interactions with the Spark service,
+ * including executing spark-shell commands.
+ */
+public class SparkService {
+
+ private final CommandExecutor executor;
+
+ public SparkService(ContainerProvider provider, String containerName) {
+ this.executor = new CommandExecutor(provider.getContainer(containerName));
+ }
+
+ /**
+ * Execute a Spark SQL command file.
+ */
+ public CommandResult executeSQLFile(String commandFile) throws Exception {
+ String sparkShellCmd = new StringBuilder()
+ .append("spark-shell --jars
").append(TestcontainersConfig.Paths.SPARK_BUNDLE)
+ .append(" --master local[2] --driver-class-path
").append(TestcontainersConfig.Paths.HADOOP_CONF_DIR)
+ .append(" --conf
spark.serializer=org.apache.spark.serializer.KryoSerializer")
+ .append(" --conf
spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog")
+ .append(" --conf
spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
+ .append(" --deploy-mode client --driver-memory 1G --executor-memory 1G
--num-executors 1")
+ // Pipe via stdin instead of `-i`. With `-i`, Spark 4's REPL boots in
+ // dumb-terminal mode (`WARN jline: Unable to create a system terminal,
+ // creating a dumb terminal`) and silently runs the script — neither
+ // input echo nor exception traces reach stdout, so success markers and
+ // failures alike are invisible to assertStdOutContains.
+ .append(" < ").append(commandFile)
+ .toString();
+
+ return executor.executeCommandString(sparkShellCmd);
+ }
+
+ /**
+ * Executes an arbitrary shell command inside the service's container.
+ * This is useful for hdfs commands, hudi-cli, etc.
+ *
+ * @param command The shell command to execute.
+ * @return The result of the command execution.
+ */
+ public CommandResult executeShellCommand(String command) throws Exception {
+ return executor.executeCommandString(command);
+ }
+}
