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 testMergingTablesWithExactSameSchema() {
        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
        assertThat(schemaDerivation.applySchemaChange(new CreateTableEvent(TABLE_2, SCHEMA)))
                .isEmpty();

        // Add column for table 1
        AddColumnEvent.ColumnWithPosition newCol1 =
                new AddColumnEvent.ColumnWithPosition(
                        new PhysicalColumn("new_col1", DataTypes.STRING(), null));
        AddColumnEvent.ColumnWithPosition newCol2 =
                new AddColumnEvent.ColumnWithPosition(
                        new PhysicalColumn("new_col2", DataTypes.STRING(), null));
        List<AddColumnEvent.ColumnWithPosition> newColumns = Arrays.asList(newCol1, newCol2);
        List<SchemaChangeEvent> derivedChangesAfterAddColumn =
                schemaDerivation.applySchemaChange(new AddColumnEvent(TABLE_1, newColumns));
        assertThat(derivedChangesAfterAddColumn).hasSize(1);
        assertThat(derivedChangesAfterAddColumn.get(0))
                .asAddColumnEvent()
                .hasTableId(MERGED_TABLE)
                .containsAddedColumns(newCol1, newCol2);
        derivedChangesAfterAddColumn.forEach(schemaManager::applyEvolvedSchemaChange);

        // Add column for table 2
        assertThat(schemaDerivation.applySchemaChange(new AddColumnEvent(TABLE_2, newColumns)))
                .isEmpty();

        // Alter column type for table 1
        ImmutableMap<String, DataType> typeMapping = ImmutableMap.of("age", DataTypes.BIGINT());
        List<SchemaChangeEvent> derivedChangesAfterAlterColumnType =
                schemaDerivation.applySchemaChange(new AlterColumnTypeEvent(TABLE_1, typeMapping));
        assertThat(derivedChangesAfterAlterColumnType).hasSize(1);
        assertThat(derivedChangesAfterAlterColumnType.get(0))
                .asAlterColumnTypeEvent()
                .hasTableId(MERGED_TABLE)
                .containsTypeMapping(typeMapping);
        derivedChangesAfterAlterColumnType.forEach(schemaManager::applyEvolvedSchemaChange);

        // Alter column type for table 2
        assertThat(
                        schemaDerivation.applySchemaChange(
                                new AlterColumnTypeEvent(TABLE_2, typeMapping)))
                .isEmpty();

        // Drop column for table 1
        List<String> droppedColumns = Arrays.asList("new_col1", "new_col2");
        assertThat(schemaDerivation.applySchemaChange(new DropColumnEvent(TABLE_1, droppedColumns)))
                .isEmpty();
        // Drop column for table 2
        assertThat(schemaDerivation.applySchemaChange(new DropColumnEvent(TABLE_2, droppedColumns)))
                .isEmpty();

        // Rename column for table 1
        Map<String, String> renamedColumns = ImmutableMap.of("name", "last_name");
        List<SchemaChangeEvent> derivedChangesAfterRenameColumn =
                schemaDerivation.applySchemaChange(new RenameColumnEvent(TABLE_1, renamedColumns));
        assertThat(derivedChangesAfterRenameColumn).hasSize(1);
        assertThat(derivedChangesAfterRenameColumn.get(0))
                .asAddColumnEvent()
                .hasTableId(MERGED_TABLE)
                .containsAddedColumns(
                        new AddColumnEvent.ColumnWithPosition(
                                new PhysicalColumn("last_name", DataTypes.STRING(), null)));
        derivedChangesAfterRenameColumn.forEach(schemaManager::applyEvolvedSchemaChange);

        // Rename column for table 2
        assertThat(
                        schemaDerivation.applySchemaChange(
                                new RenameColumnEvent(TABLE_2, renamedColumns)))
                .isEmpty();
    }