hudi-agent commented on code in PR #19720: URL: https://github.com/apache/hudi/pull/19720#discussion_r3845821353
########## hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestConcurrentInsertOverwriteSamePartition.java: ########## @@ -0,0 +1,171 @@ +/* + * 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.client; + +import org.apache.hudi.client.transaction.SimpleConcurrentFileWritesConflictResolutionStrategy; +import org.apache.hudi.client.transaction.lock.InProcessLockProvider; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.model.HoodieReplaceCommitMetadata; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.model.WriteConcurrencyMode; +import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.marker.MarkerType; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.testutils.HoodieTestUtils; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.config.HoodieCleanConfig; +import org.apache.hudi.config.HoodieLockConfig; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieWriteConflictException; +import org.apache.hudi.table.HoodieSparkTable; +import org.apache.hudi.testutils.HoodieClientTestBase; + +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + +import static org.apache.hudi.common.model.HoodieRecord.RECORD_KEY_METADATA_FIELD; +import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH; +import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Two writers running INSERT_OVERWRITE of the same partition concurrently under OCC. When the partition is + * empty at planning time neither replaces any file id, so the file-id based conflict check alone would let + * both commit and leave every record in the partition twice. + */ +public class TestConcurrentInsertOverwriteSamePartition extends HoodieClientTestBase { + + private static final String TARGET_PARTITION = DEFAULT_FIRST_PARTITION_PATH; + private static final String OTHER_PARTITION = DEFAULT_SECOND_PARTITION_PATH; + private static final int RECORDS_PER_WRITE = 100; + + @Override + public SparkRDDWriteClient getHoodieWriteClient(HoodieWriteConfig cfg) { + return new SparkRDDWriteClient(context, cfg); + } + + /** A retried load whose previous attempt is still running: both overwrites planned against an empty partition. */ + @Test + public void testConcurrentInsertOverwriteOfEmptyPartitionIsRejected() throws Exception { + HoodieWriteConfig cfg = occWriteConfig(); + // Seed the table with data in an unrelated partition so it is an existing, non-empty table. + try (SparkRDDWriteClient client = getHoodieWriteClient(cfg)) { + String seedTime = client.startCommit(); + client.commit(seedTime, client.insert(jsc.parallelize( + dataGen.generateInsertsForPartition(seedTime, RECORDS_PER_WRITE, OTHER_PARTITION), 1), seedTime)); + } + + // Two writers, same payload, same target partition: + // both start (requested) before either finishes. + String replaceAction = HoodieTimeline.REPLACE_COMMIT_ACTION; + String firstInstant; + String secondInstant; + + try (SparkRDDWriteClient writer1 = getHoodieWriteClient(cfg); + SparkRDDWriteClient writer2 = getHoodieWriteClient(cfg)) { + firstInstant = writer1.startCommit(replaceAction); + secondInstant = writer2.startCommit(replaceAction); + List<HoodieRecord> payload = dataGen.generateInsertsForPartition(firstInstant, RECORDS_PER_WRITE, TARGET_PARTITION); + + HoodieWriteResult result1 = writer1.insertOverwrite(jsc.parallelize(payload, 1), firstInstant); + HoodieWriteResult result2 = writer2.insertOverwrite(jsc.parallelize(payload, 1), secondInstant); + + assertTrue(writer1.commit(firstInstant, result1.getWriteStatuses(), Option.empty(), + replaceAction, result1.getPartitionToReplaceFileIds()), "first overwrite commits"); + assertTrue(result1.getPartitionToReplaceFileIds().get(TARGET_PARTITION).isEmpty() + && result2.getPartitionToReplaceFileIds().get(TARGET_PARTITION).isEmpty(), "both planned against an empty partition"); + assertThrows(HoodieWriteConflictException.class, () -> writer2.commit(secondInstant, result2.getWriteStatuses(), + Option.empty(), replaceAction, result2.getPartitionToReplaceFileIds()), "second overwrite of the same partition is rejected"); + } + + HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(storageConf, basePath); + HoodieTimeline replaceTimeline = metaClient.getActiveTimeline().getCompletedReplaceTimeline(); + assertEquals(Arrays.asList(firstInstant), Review Comment: 🤖 nit: `Collections.singletonList(firstInstant)` would be slightly more idiomatic here than `Arrays.asList` with a single element — makes the intent (exactly one element) immediately clear. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestConcurrentInsertOverwritePartition.scala: ########## @@ -0,0 +1,160 @@ +/* + * 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.spark.sql.hudi.dml.insert + +import org.apache.hudi.common.model.{HoodieReplaceCommitMetadata, WriteOperationType} +import org.apache.hudi.common.table.HoodieTableMetaClient +import org.apache.hudi.exception.HoodieWriteConflictException +import org.apache.hudi.hadoop.fs.HadoopFSUtils + +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase + +import java.util.concurrent.{CountDownLatch, Executors, TimeUnit} + +import scala.collection.JavaConverters._ +import scala.util.Try + +/** + * Two SQL sessions run INSERT OVERWRITE of the same, still empty, partition of a MOR table at the same time, + * the way a retried daily load does when the previous attempt is still running. + */ +class TestConcurrentInsertOverwritePartition extends HoodieSparkSqlTestBase { + + private val targetPartition = "2026-08-04" + private val seededPartition = "2026-08-03" + private val rowsPerWriter = 200 + + test("Concurrent INSERT OVERWRITE of the same empty partition under OCC") { + withTempDir { tmp => + val tableName = generateTableName + val basePath = s"${tmp.getCanonicalPath}/$tableName" + spark.sql( + s""" + | create table $tableName ( + | rank_source_id int, + | locode string, + | device string, + | normalized_query string, + | serp_item_position int, + | url string, + | modeled_ingested_at long, + | day string + | ) using hudi + | partitioned by (day) + | tblproperties ( + | type = 'mor', + | primaryKey = 'rank_source_id,locode,device,normalized_query,serp_item_position', + | preCombineField = 'modeled_ingested_at', + | hoodie.datasource.write.hive_style_partitioning = 'true', + | hoodie.metadata.enable = 'true' + | ) + | location '$basePath' + """.stripMargin) + // Seed an unrelated partition so the table has a completed commit older than both overwrites. On a + // brand-new table the loser's instant can sort below the first completed commit and its uncleaned + // files would then be read as if they belonged to an archived commit, hiding the outcome under test. + spark.sql( + s""" + | insert into $tableName partition (day = '$seededPartition') + | values (0, 'US', 'desktop', 'seed', 1, 'https://example.com/seed', 0L) + """.stripMargin) + // Source data lives in one file so each INSERT OVERWRITE runs the gate UDF in exactly one task. + val sourceTable = generateTableName + spark.sql(s"create table $sourceTable using parquet as select id from range(0, $rowsPerWriter, 1, 1)") + + val pool = Executors.newFixedThreadPool(2) + val writers = try { + Seq("writer_1", "writer_2").map(name => pool.submit(() => Try { + runInsertOverwrite(name, tableName, sourceTable) + })).map(_.get(5, TimeUnit.MINUTES)) + } finally { + pool.shutdownNow() + } + assertWriterOutcome(writers, basePath, tableName) + } + } + + private def runInsertOverwrite(name: String, tableName: String, sourceTable: String): String = { + val session = spark.newSession() + session.udf.register("wait_for_other_writer", (id: Long) => { + TestConcurrentInsertOverwritePartition.bothWritersPlanned.countDown() + // Release only once both statements have started their write stage, i.e. both have + // initialised their table view and planned the overwrite against an empty partition. + assert(TestConcurrentInsertOverwritePartition.bothWritersPlanned.await(2, TimeUnit.MINUTES), + "the other writer never reached its write stage") + id + }) + occSqlConf.foreach { case (k, v) => session.sql(s"set $k=$v") } + session.sql( + s""" + | insert overwrite table $tableName partition (day = '$targetPartition') + | select cast(id as int), 'US', 'desktop', concat('q', id), 1, concat('https://example.com/', id), + | wait_for_other_writer(id) as modeled_ingested_at + | from $sourceTable + """.stripMargin) + name + } + + private def assertWriterOutcome(writers: Seq[Try[String]], basePath: String, tableName: String): Unit = { + val metaClient = HoodieTableMetaClient.builder().setConf(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf())).setBasePath(basePath).build() + val completedOverwrites = metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.asScala + .map(i => (i, metaClient.getActiveTimeline.readReplaceCommitMetadata(i))) + .filter(_._2.getOperationType == WriteOperationType.INSERT_OVERWRITE) + + val rowCount = spark.sql(s"select count(*) from $tableName where day = '$targetPartition'").head().getLong(0) + val keyCount = spark.sql( + s"select count(distinct rank_source_id, locode, device, normalized_query, serp_item_position) from $tableName where day = '$targetPartition'") + .head().getLong(0) + + println(s"overwrites committed=${completedOverwrites.map { case (i, md) => s"$i replaced=${md.getPartitionToReplaceFileIds}" }} " + Review Comment: 🤖 nit: could you swap `println` for a logger (e.g. `log.info(...)`)? Raw `println` gets swallowed by test frameworks in CI and mixed with unrelated output, making it harder to trace a failure. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
