private SplitEnumerator<IcebergSourceSplit, IcebergEnumeratorState> createEnumerator(
            SplitEnumeratorContext<IcebergSourceSplit> enumContext,
            @Nullable IcebergEnumeratorState enumState) {
        SplitAssigner assigner;
        if (enumState == null) {
            assigner = assignerFactory.createAssigner();
        } else {
            LOG.info(
                    "Iceberg source restored {} splits from state for table {}",
                    enumState.pendingSplits().size(),
                    lazyTable().name());
            assigner = assignerFactory.createAssigner(enumState.pendingSplits());
        }

        if (scanContext.isStreaming()) {
            ContinuousSplitPlanner splitPlanner =
                    new ContinuousSplitPlannerImpl(tableLoader.clone(), scanContext, planningThreadName());
            return new ContinuousIcebergEnumerator(
                    enumContext, assigner, scanContext, splitPlanner, enumState);
        } else {
            List<IcebergSourceSplit> splits = planSplitsForBatch(planningThreadName());
            assigner.onDiscoveredSplits(splits);
            return new StaticIcebergEnumerator(enumContext, assigner);
        }
    }
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/IcebergSource.java