@ParameterizedTest(name = "Test-NewCatalog-Load-HistoricalTable[withPrimaryKey={0}]")
    @ValueSource(booleans = {false, true})
    public void testNewCatalogLoadHistorical(boolean primaryKey) {
      TableBuilder builder =
          historicalCatalog.newTableBuilder(tableIdentifier, schema).withPartitionSpec(spec);
      if (primaryKey) {
        builder.withPrimaryKeySpec(keySpec);
      }
      // create historical table
      MixedTable historicalTable = builder.create();
      // write and commit historical table
      List<Record> expectedResult = writeTestData(historicalTable);

      // load table base on rest catalog
      MixedTable restTable = catalog.loadTable(tableIdentifier);
      List<Record> loadedRecords =
          MixedDataTestHelpers.readTable(restTable, Expressions.alwaysTrue());

      // assert rest-based table scan right records.
      Assertions.assertEquals(expectedResult.size(), loadedRecords.size());

      // write through rest-based table
      List<Record> results = writeTestData(restTable, null, newChangeAdded, newChangeDelete);

      // read data through historical table
      List<Record> hisScanResult =
          MixedDataTestHelpers.readTable(historicalTable, Expressions.alwaysTrue());
      Assertions.assertEquals(hisScanResult.size(), results.size());

      // read data through rest-based table
      List<Record> restScanResult =
          MixedDataTestHelpers.readTable(restTable, Expressions.alwaysTrue());
      Assertions.assertEquals(results.size(), restScanResult.size());

      // read data through historical table.
      List<Record> scanHistorical =
          MixedDataTestHelpers.readTable(historicalTable, Expressions.alwaysTrue());
      Assertions.assertEquals(results.size(), scanHistorical.size());

      // TODO: there is bug in unkeyed-table.location.
      String location = getTableMetadata(tableIdentifier).getTableLocation();
      AuthenticatedFileIO io = historicalTable.io();
      // drop table through rest-catalog
      catalog.dropTable(tableIdentifier, true);
      Assertions.assertTrue(catalog.listTables(database).isEmpty());
      Assertions.assertTrue(historicalCatalog.listTables(database).isEmpty());
      // table dir has been purged
      Assertions.assertFalse(io.exists(location));
    }
The purpose of these tests is to test the compatibility of historical versions of connectors.
Mixed-iceberg tables created by old connectors can be accessed with new connectors, but tables
created by new connectors cannot be accessed with old versions.
amoro-ams/src/test/java/org/apache/amoro/server/TestInternalMixedCatalogService.java