This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit adebc4c95682b7b481ab61b8ed97d549118a3353 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. (cherry picked from commit 3015eb944083272459040dc5da5755e683674cb5) --- .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 fb7dd760c505..851b53e78618 100644 --- a/.github/workflows/bot.yml +++ b/.github/workflows/bot.yml @@ -1251,6 +1251,82 @@ jobs: rm -f $GITHUB_WORKSPACE/$SPARK_ARCHIVE mvn verify $SCALA_PROFILE -D"$SPARK_PROFILE" -Pintegration-tests -pl !hudi-flink-datasource/hudi-flink $MVN_ARGS + 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 strategy: 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 4910031b5a47..f2f2633847ba 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); + } +}
