sanqingleo created HUDI-6675:
--------------------------------
Summary: InsertOverwrite will delete the whole table
Key: HUDI-6675
URL: https://issues.apache.org/jira/browse/HUDI-6675
Project: Apache Hudi
Issue Type: Bug
Components: cleaning
Affects Versions: 0.13.0, 0.11.1
Environment: hudi 0.11 both 0.13.
spark 3.4
Reporter: sanqingleo
Attachments: image-2023-08-10-10-35-02-798.png,
image-2023-08-10-10-37-05-339.png
h1. Abstract
when I use inset_overwrite feature both in spark sql and api, It's will clean
the whole table when it's not partition table
then throw this exception
!image-2023-08-10-10-37-05-339.png!
h1. Version
# hudi 0.11 both 0.13.
# spark 3.4
h1. Bug Position
org.apache.hudi.table.action.clean.CleanActionExecutor#deleteFileAndGetResult
!image-2023-08-10-10-35-02-798.png!
h1. How to recurrent
Need to run 4 times, fourth time will trigger clean action.
0.11, both sql and api
0.13 just api
{code:java}
import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
object InsertOverwriteTest {
def main(array: Array[String]): Unit = {
val spark = SparkSession.builder()
.appName("TestInsertOverwrite")
.master("local[4]")
.config("spark.sql.extensions",
"org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.config("spark.sql.catalog.spark_catalog"
,"org.apache.spark.sql.hudi.catalog.HoodieCatalog")
.getOrCreate()
spark.conf.set("hoodie.index.type", "BUCKET")
spark.conf.set("hoodie.storage.layout.type", "BUCKET")
spark.conf.set("HADOOP_USER_NAME", "parallels")
System.setProperty("HADOOP_USER_NAME", "parallels")
var seq = List(
Row("uuid_01", "27", "2022-09-23", "par_01"),
Row("uuid_02", "21", "2022-09-23", "par_02"),
Row("uuid_03", "23", "2022-09-23", "par_04"),
Row("uuid_04", "24", "2022-09-23", "par_02"),
Row("uuid_05", "26", "2022-09-23", "par_01"),
Row("uuid_06", "20", "2022-09-23", "par_03"),
)
var rdd = spark.sparkContext.parallelize(seq)
var structType: StructType = StructType(Array(
StructField("uuid", DataTypes.StringType, nullable = true),
StructField("age", DataTypes.StringType, nullable = true),
StructField("ts", DataTypes.StringType, nullable = true),
StructField("par", DataTypes.StringType, nullable = true)
))
var df1 = spark.createDataFrame(rdd, structType)
.createOrReplaceTempView("compact_test_num")
var df: DataFrame = spark.sql(" select uuid, age, ts, par from
compact_test_num limit 10")
df.write.format("org.apache.hudi")
.option(RECORDKEY_FIELD.key, "uuid")
.option(PRECOMBINE_FIELD.key, "ts")
// .option(PARTITIONPATH_FIELD.key(), "par")
.option("hoodie.table.keygenerator.class",
"org.apache.hudi.keygen.NonpartitionedKeyGenerator")
.option(KEYGENERATOR_CLASS_NAME.key,
"org.apache.hudi.keygen.NonpartitionedKeyGenerator")
// .option(KEYGENERATOR_CLASS_NAME.key,
"org.apache.hudi.keygen.ComplexKeyGenerator")
.option(OPERATION.key, INSERT_OVERWRITE_OPERATION_OPT_VAL)
.option(TABLE_TYPE.key, COW_TABLE_TYPE_OPT_VAL)
.option("hoodie.metadata.enable", "false")
.option("hoodie.index.type", "BUCKET")
.option("hoodie.bucket.index.hash.field", "uuid")
.option("hoodie.bucket.index.num.buckets", "2")
.option("hoodie.storage.layout.type", "BUCKET")
.option("hoodie.storage.layout.partitioner.class",
"org.apache.hudi.table.action.commit.SparkBucketIndexPartitioner")
.option("hoodie.table.name", "cow_20230801_012")
.option("hoodie.upsert.shuffle.parallelism", "2")
.option("hoodie.insert.shuffle.parallelism", "2")
.option("hoodie.delete.shuffle.parallelism", "2")
.option("hoodie.clean.max.commits", "2")
.option("hoodie.cleaner.commits.retained", "2")
.option("hoodie.datasource.write.hive_style_partitioning", "true")
.mode(SaveMode.Append)
.save("hdfs://bigdata01:9000/hudi_test/cow_20230801_012")
}
}
{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
