public List<SchemaChangeEvent> applySchemaChange(SchemaChangeEvent schemaChangeEvent) {
        List<SchemaChangeEvent> events = new ArrayList<>();
        TableId originalTable = schemaChangeEvent.tableId();
        boolean noRouteMatched = true;

        for (Tuple3<Selectors, String, String> route : routes) {
            // Check routing table
            if (!route.f0.isMatch(originalTable)) {
                continue;
            }

            noRouteMatched = false;

            // Matched a routing rule
            TableId derivedTable = resolveReplacement(originalTable, route);
            Set<TableId> originalTables =
                    derivationMapping.computeIfAbsent(derivedTable, t -> new HashSet<>());
            originalTables.add(originalTable);

            if (originalTables.size() == 1) {
                // single source mapping, replace the table ID directly
                SchemaChangeEvent derivedSchemaChangeEvent =
                        ChangeEventUtils.recreateSchemaChangeEvent(schemaChangeEvent, derivedTable);
                events.add(derivedSchemaChangeEvent);
            } else {
                // multiple source mapping (merging tables)
                Schema derivedTableSchema =
                        schemaManager.getLatestEvolvedSchema(derivedTable).get();
                events.addAll(
                        Objects.requireNonNull(
                                SchemaChangeEventVisitor.visit(
                                        schemaChangeEvent,
                                        addColumnEvent ->
                                                handleAddColumnEvent(
                                                        addColumnEvent,
                                                        derivedTableSchema,
                                                        derivedTable),
                                        alterColumnTypeEvent ->
                                                handleAlterColumnTypeEvent(
                                                        alterColumnTypeEvent,
                                                        derivedTableSchema,
                                                        derivedTable),
                                        createTableEvent ->
                                                handleCreateTableEvent(
                                                        createTableEvent,
                                                        derivedTableSchema,
                                                        derivedTable),
                                        dropColumnEvent ->
                                                Collections.emptyList(), // Column drop shouldn't be
                                        // spread to route
                                        // destination.
                                        dropTableEvent ->
                                                Collections.emptyList(), // Table drop shouldn't be
                                        // spread to route
                                        // destination.
                                        renameColumnEvent ->
                                                handleRenameColumnEvent(
                                                        renameColumnEvent,
                                                        derivedTableSchema,
                                                        derivedTable),
                                        truncateTableEvent ->
                                                Collections.emptyList() // // Table truncation
                                        // shouldn't be spread to route
                                        // destination.
                                        )));
            }
        }

        if (noRouteMatched) {
            // No routes are matched, leave it as-is
            return Collections.singletonList(schemaChangeEvent);
        } else {
            return events;
        }
    }
@Test
    void testIncompatibleTypes() {
        SchemaManager schemaManager = new SchemaManager();
        SchemaDerivation schemaDerivation =
                new SchemaDerivation(schemaManager, ROUTES, new HashMap<>());
        // Create table 1
        List<SchemaChangeEvent> derivedChangesAfterCreateTable =
                schemaDerivation.applySchemaChange(new CreateTableEvent(TABLE_1, SCHEMA));
        assertThat(derivedChangesAfterCreateTable).hasSize(1);
        assertThat(derivedChangesAfterCreateTable.get(0))
                .asCreateTableEvent()
                .hasTableId(MERGED_TABLE)
                .hasSchema(SCHEMA);
        derivedChangesAfterCreateTable.forEach(schemaManager::applyEvolvedSchemaChange);

        // Create table 2
        assertThatThrownBy(
                        () ->
                                schemaDerivation.applySchemaChange(
                                        new CreateTableEvent(TABLE_2, INCOMPATIBLE_SCHEMA)))
                .isInstanceOf(IllegalStateException.class)
                .hasMessage("Incompatible types: \"INT\" and \"STRING\"");
    }