object MergeTreePartSplit {
  def apply(
      name: String,
      dirName: String,
      targetNode: String,
      start: Long,
      length: Long,
      bytesOnDisk: Long
  ): MergeTreePartSplit = {
    // Ref to org.apache.spark.sql.delta.files.TahoeFileIndex.absolutePath
    val uriDecodeName = new Path(new URI(name)).toString
    val uriDecodeDirName = new Path(new URI(dirName)).toString
    new MergeTreePartSplit(uriDecodeName, uriDecodeDirName, targetNode, start, length, bytesOnDisk)
  }
}

test("test partitioned with escaped characters") {

  val schema = StructType(
    Seq(
      StructField.apply("id", IntegerType, nullable = true),
      StructField.apply("escape", StringType, nullable = true),
      StructField.apply("bucket/col", StringType, nullable = true),
      StructField.apply("part=col1", DateType, nullable = true),
      StructField.apply("part_col2", StringType, nullable = true)
    ))

  val data: Seq[Row] = Seq(
    Row(1, "=", "00000", Date.valueOf("2024-01-01"), "2024=01/01"),
    Row(2, "/", "00000", Date.valueOf("2024-01-01"), "2024=01/01"),
    Row(3, "#", "00000", Date.valueOf("2024-01-01"), "2024#01:01"),
    Row(4, ":", "00001", Date.valueOf("2024-01-02"), "2024#01:01"),
    Row(5, "\\", "00001", Date.valueOf("2024-01-02"), "2024\\01\u000101"),
    Row(6, "\u0001", "000001", Date.valueOf("2024-01-02"), "2024\\01\u000101"),
    Row(7, "", "000002", null, null)
  )

  val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
  df.createOrReplaceTempView("origin_table")
  spark.sql("select * from origin_table").show()

  nativeWrite {
    format =>
      val table_name = table_name_template.format(format)
      spark.sql(s"drop table IF EXISTS $table_name")
      writeAndCheckRead("origin_table", table_name, schema.fieldNames.map(f => s"`$f`")) {
        _ =>
          spark
            .table("origin_table")
            .write
            .format(format)
            .partitionBy("part=col1", "part_col2")
            .bucketBy(2, "bucket/col")
            .saveAsTable(table_name)
      }

      val table_name_vanilla = table_name_vanilla_template.format(format)
      spark.sql(s"drop table IF EXISTS $table_name_vanilla")
      withSQLConf((GlutenConfig.NATIVE_WRITER_ENABLED.key, "false")) {
        withNativeWriteCheck(checkNative = false) {
          spark
            .table("origin_table")
            .write
            .format(format)
            .partitionBy("part=col1", "part_col2")
            .bucketBy(2, "bucket/col")
            .saveAsTable(table_name_vanilla)
        }
        compareWriteFilesSignature(format, table_name, table_name_vanilla, "sum(id)")
      }
  }
}

test("test bucketed by constant") {