apache/seatunnel · error · UnsupportedOperationException

Multiple input tables are not supported in the current…

Error message

Multiple input tables are not supported in the current version

What it means

SparkAbstractPluginExecuteProcessor.fromSourceTable resolves a plugin's declared upstream input tables (plugin_input). If the plugin declares more than one input table identifier, the Spark Dataset-based execution path — which supports only single-input operators — throws UnsupportedOperationException. Multi-input joins on Spark are not implemented in this version.

Solutions

1. Reduce plugin_input to a single table for Spark jobs
2. Restructure the DAG so multi-input transforms run on Zeta engine instead of Spark
3. Pre-merge upstream tables with a supported single-input transform chain
4. Split the job into multiple stages writing intermediate results

Example fix

// before
plugin_input = ["table_a", "table_b"]
// after
plugin_input = "table_a"

if (pluginInputIdentifiers != null && pluginInputIdentifiers.size() > 1) {
    throw new IllegalArgumentException("Spark engine supports only one plugin_input table");
}

boolean isSingleInput(List<String> inputs) {
    return inputs == null || inputs.size() <= 1;
}

try {
    runSparkJob();
} catch (UnsupportedOperationException e) {
    if (e.getMessage().contains("Multiple input tables")) {
        log.error("Run this DAG on Zeta engine or reduce plugin_input to one table");
    }
}

When it happens

Trigger: A transform/plugin config sets plugin_input to a list with more than one table name, e.g. plugin_input = ["table_a", "table_b"], and the job runs on the Spark engine.

Common situations: Sharing one config file across engines (Zeta/Flink support multi-input, Spark does not); users trying to implement join-type transforms on Spark; copy-paste of Flink configs.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• All candidate sink tables were skipped in Spark starter.
• All candidate sink tables were skipped in Spark starter.
• Skip failed sink table in Spark starter
• Skip failed sink table in Spark starter
• Some sink tables were skipped in Spark starter.

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/83b7a5cfdc30bcef. Report an issue: GitHub.

Appendix: source

Thrown at seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/execution/SparkAbstractPluginExecuteProcessor.java:88

    protected void registerInputTempView(Config pluginConfig, Dataset<Row> dataStream) {
        ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(pluginConfig);
        if (readonlyConfig.getOptional(PLUGIN_OUTPUT).isPresent()) {
            String tableName = readonlyConfig.get(PLUGIN_OUTPUT);
            registerTempView(tableName, dataStream);
        }
    }

    protected Optional<DatasetTableInfo> fromSourceTable(
            Config pluginConfig,
            SparkRuntimeEnvironment sparkRuntimeEnvironment,
            List<DatasetTableInfo> upstreamDataStreams) {
        List<String> pluginInputIdentifiers =
                ReadonlyConfig.fromConfig(pluginConfig).get(PLUGIN_INPUT);
        if (pluginInputIdentifiers == null || pluginInputIdentifiers.isEmpty()) {
            return Optional.empty();
        }
        if (pluginInputIdentifiers.size() > 1) {
            throw new UnsupportedOperationException(
                    "Multiple input tables are not supported in the current version");
        }
        String pluginInputIdentifier = pluginInputIdentifiers.get(0);
        DatasetTableInfo datasetTableInfo =
                upstreamDataStreams.stream()
                        .filter(info -> pluginInputIdentifier.equals(info.getTableName()))
                        .findFirst()
                        .orElseThrow(
                                () ->
                                        new SeaTunnelException(
                                                String.format(
                                                        "table %s not found",
                                                        pluginInputIdentifier)));
        return Optional.of(
                new DatasetTableInfo(
                        sparkRuntimeEnvironment
                                .getSparkSession()
                                .read()

View on GitHub (pinned to cf67b549a7)