@DisplayName("TestSQL: INSERT INTO table with upsert enabled")
  @ParameterizedTest
  @MethodSource
  public void testUpsert(
      TableFormat format,
      Schema schema,
      PrimaryKeySpec keySpec,
      PartitionSpec ptSpec,
      FileFormat fileFormat) {
    MixedTable table =
        createTarget(
            schema,
            tableBuilder ->
                tableBuilder
                    .withPrimaryKeySpec(keySpec)
                    .withProperty(TableProperties.UPSERT_ENABLED, "true")
                    .withProperty(TableProperties.CHANGE_FILE_FORMAT, fileFormat.name())
                    .withProperty(TableProperties.BASE_FILE_FORMAT, fileFormat.name())
                    .withPartitionSpec(ptSpec));
    createViewSource(schema, source);

    TestTableUtil.writeToBase(table, base);
    sql("INSERT INTO " + target() + " SELECT * FROM " + source());

    List<Record> expects;
    if (keySpec.primaryKeyExisted()) {
      expects = ExpectResultUtil.upsertResult(base, source, r -> r.get(0, Integer.class));
    } else {
      expects = Lists.newArrayList(base);
      expects.addAll(source);
    }

    table.refresh();
    List<Record> results = TestTableUtil.tableRecords(table);
    DataComparator.build(expects, results).ignoreOrder(comparator).assertRecordsEqual();

    if (table.isKeyedTable()) {
      List<Record> deletes =
          ExpectResultUtil.upsertDeletes(base, source, r -> r.get(0, Integer.class));

      List<Record> expectChanges =
          deletes.stream()
              .map(
                  r ->
                      TestTableUtil.extendMetadataValue(
                          r, MetadataColumns.CHANGE_ACTION_FIELD, ChangeAction.DELETE.name()))
              .collect(Collectors.toList());

      source.stream()
          .map(
              r ->
                  TestTableUtil.extendMetadataValue(
                      r, MetadataColumns.CHANGE_ACTION_FIELD, ChangeAction.INSERT.name()))
          .forEach(expectChanges::add);

      List<Record> changes = TestTableUtil.changeRecordsWithAction(table.asKeyedTable());

      DataComparator.build(expectChanges, changes)
          .ignoreOrder(pkComparator.thenComparing(changeActionComparator))
          .assertRecordsEqual();
    }
  }
test 1. upsert not enabled test 2. upsert enabled test 3. upsert source duplicate check test 4.
upsert optimize write
amoro-format-mixed/amoro-mixed-spark/v3.3/amoro-mixed-spark-3.3/src/test/java/org/apache/amoro/spark/test/suites/sql/TestInsertIntoSQL.java