@Override
  public DynamicTableSource createDynamicTableSource(Context context) {
    CatalogTable catalogTable = context.getCatalogTable();
    ObjectIdentifier identifier = context.getObjectIdentifier();
    ObjectPath objectPath;

    FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
    Configuration options = (Configuration) helper.getOptions();

    InternalCatalogBuilder actualBuilder = internalCatalogBuilder;
    String actualCatalogName = internalCatalogName;

    // It denotes create table by ddl 'connector' option, not through catalog.db.tableName
    if (actualBuilder == null || actualCatalogName == null) {
      actualCatalogName = options.get(MixedFormatValidator.MIXED_FORMAT_CATALOG);
      Preconditions.checkNotNull(
          actualCatalogName,
          String.format("%s should be set", MixedFormatValidator.MIXED_FORMAT_CATALOG.key()));
      String metastoreUrl = options.get(CatalogFactoryOptions.METASTORE_URL);
      actualBuilder =
          InternalCatalogBuilder.builder()
              .metastoreUrl(metastoreUrl)
              .catalogName(actualCatalogName)
              .properties(options.toMap());
    }

    if (options.containsKey(MixedFormatValidator.MIXED_FORMAT_DATABASE.key())
        && options.containsKey(MixedFormatValidator.MIXED_FORMAT_TABLE.key())) {
      objectPath =
          new ObjectPath(
              options.get(MixedFormatValidator.MIXED_FORMAT_DATABASE),
              options.get(MixedFormatValidator.MIXED_FORMAT_TABLE));
    } else {
      objectPath = new ObjectPath(identifier.getDatabaseName(), identifier.getObjectName());
    }
    MixedFormatTableLoader tableLoader =
        createTableLoader(objectPath, actualCatalogName, actualBuilder, options.toMap());
    MixedTable mixedTable = MixedFormatUtils.loadMixedTable(tableLoader);

    Configuration confWithAll = Configuration.fromMap(mixedTable.properties());

    ScanTableSource mixedFormatDynamicSource;

    String readMode =
        PropertyUtil.propertyAsString(
            mixedTable.properties(),
            MixedFormatValidator.MIXED_FORMAT_READ_MODE,
            MixedFormatValidator.MIXED_READ_MODE_DEFAULT);

    boolean dimTable =
        CompatibleFlinkPropertyUtil.propertyAsBoolean(
            mixedTable.properties(),
            MixedFormatValidator.DIM_TABLE_ENABLE.key(),
            MixedFormatValidator.DIM_TABLE_ENABLE.defaultValue());

    TableSchema tableSchema;
    if (!dimTable) {
      tableSchema =
          org.apache.amoro.flink.FlinkSchemaUtil.getPhysicalSchema(catalogTable.getSchema());
    } else {
      tableSchema =
          org.apache.amoro.flink.FlinkSchemaUtil.getPhysicalSchemaForDimTable(
              catalogTable.getSchema());
    }

    switch (readMode) {
      case MixedFormatValidator.MIXED_FORMAT_READ_FILE:
        boolean batchMode = context.getConfiguration().get(RUNTIME_MODE).equals(BATCH);
        LOG.info("Building a file reader in {} runtime mode", batchMode ? "batch" : "streaming");
        mixedFormatDynamicSource =
            new MixedFormatFileSource(tableLoader, tableSchema, mixedTable, confWithAll, batchMode);
        break;
      case MixedFormatValidator.MIXED_FORMAT_READ_LOG:
      default:
        Preconditions.checkArgument(
            CompatiblePropertyUtil.propertyAsBoolean(
                mixedTable.properties(), ENABLE_LOG_STORE, ENABLE_LOG_STORE_DEFAULT),
            String.format("Read log should enable %s at first", ENABLE_LOG_STORE));
        mixedFormatDynamicSource = createLogSource(mixedTable, context, confWithAll);
    }

    return generateDynamicTableSource(
        identifier.getObjectName(), mixedFormatDynamicSource, mixedTable, tableLoader);
  }
A factory generates {@link MixedFormatDynamicSource} and {@link MixedFormatDynamicSink}
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/MixedDynamicTableFactory.java