class TestFlinkUuidType extends CatalogTestBase {

  private static final String TABLE_NAME = "test_table";
  private Table icebergTable;
  @TempDir private Path warehouseDir;

  @Parameters(name = "catalogName={0}, baseNamespace={1}")
  protected static List<Object[]> parameters() {
    return Arrays.asList(
        // For now hive metadata is not supported variant, so we only test hadoop catalog
        new Object[] {"testhadoop", Namespace.empty()},
        new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1")});
  }

  @Override
  @BeforeEach
  public void before() {
    super.before();
    sql("CREATE DATABASE %s", flinkDatabase);
    sql("USE CATALOG %s", catalogName);
    sql("USE %s", DATABASE);

    Schema schema =
            new Schema(
                    Types.NestedField.required(1, "id", Types.IntegerType.get()),
                    Types.NestedField.optional(2, "data", Types.UUIDType.get()));
    icebergTable =
            validationCatalog.createTable(
                    TableIdentifier.of(icebergNamespace, TABLE_NAME),
                    schema,
                    PartitionSpec.unpartitioned(),
                    ImmutableMap.of(
                            "format-version",
                            "3",
                            "write.format.default",
                            FileFormat.AVRO.name()));
  }

  @TestTemplate
  public void testInsertUuidFromFlink() throws Exception {
    UUID expectedUuid = UUID.fromString("123e4567-e89b-12d3-a456-426614174000");
    sql(
            "INSERT INTO %s VALUES (1, CAST(X'%s' AS BINARY(16)))",
            TABLE_NAME,
            "123e4567e89b12d3a456426614174000");

    List<Record> records = SimpleDataUtil.tableRecords(icebergTable);
    assertThat(records).hasSize(1);
    assertThat(records.get(0).getField("data")).isInstanceOf(UUID.class).isEqualTo(expectedUuid);
  }

  @TestTemplate
  public void testReadUuidFromFlink() throws Exception {
    UUID expectedUuid = UUID.fromString("0f8fad5b-d9cb-469f-a165-70867728950e");
    ImmutableList.Builder<Record> builder = ImmutableList.builder();
    builder.add(GenericRecord.create(icebergTable.schema()).copy("id", 1, "data", expectedUuid));
    new GenericAppenderHelper(icebergTable, FileFormat.PARQUET, warehouseDir)
            .appendToTable(builder.build());
    icebergTable.refresh();

    List<GenericRowData> genericRowData = Lists.newArrayList();
    try (CloseableIterable<CombinedScanTask> combinedScanTasks =
                 icebergTable.newScan().planTasks()) {
      for (CombinedScanTask combinedScanTask : combinedScanTasks) {
        try (DataIterator<RowData> dataIterator =
                     ReaderUtil.createDataIterator(
                             combinedScanTask, icebergTable.schema(), icebergTable.schema())) {
          while (dataIterator.hasNext()) {
            GenericRowData rowData = (GenericRowData) dataIterator.next();
            genericRowData.add(rowData);
          }
        }
      }
    }

    assertThat(genericRowData).hasSize(1);
    assertThat(genericRowData.get(0).getField(1)).isInstanceOf(byte[].class);
    byte[] uuidBytes = (byte[]) genericRowData.get(0).getField(1);
    assertThat(uuidBytes).hasSize(16);

    ByteBuffer bb = ByteBuffer.wrap(uuidBytes);
    UUID actualUuid = new UUID(bb.getLong(), bb.getLong());
    assertThat(actualUuid).isEqualTo(expectedUuid);
  }
}