public static Transaction beginReplace(
            File temp,
            String name,
            Schema schema,
            PartitionSpec spec,
            SortOrder sortOrder,
            Map<String, String> properties,
            TestTableOperations ops) {
        TableMetadata current = ops.current();
        TableMetadata metadata;
        if (current != null) {
            metadata = current.buildReplacement(schema, spec, sortOrder, current.location(), properties);
            return Transactions.replaceTableTransaction(name, ops, metadata);
        } else {
            metadata = newTableMetadata(schema, spec, sortOrder, temp.toString(), properties);
            return Transactions.createTableTransaction(name, ops, metadata);
        }
    }
Copy from iceberg-flink:iceberg-flink-1.15:1.1.x
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/test/java/org/apache/inlong/sort/iceberg/sink/util/TestTables.java