Flink的 CTAS 目标库为starrocks，为什么报错呢？

Flink的 CTAS 目标库为starrocks，为什么报错呢？语句时照抄 https://help.aliyun.com/zh/emr/emr-on-ecs/user-guide/use-the-ctas-and-cdas-statements-of-realtime-compute-for-apache-flink-to-synchronize-data-from-an-apsaradb-rds-for-mysql-instance-to-a-starrocks-cluster?spm=a2c4g.11186623.0.0.30127e8dDcfzkb 的这里
报错：org.apache.flink.table.api.ValidationException: SQL validation failed. CREATE TABLE ... AS TABLE ... statement requires target catalog 'vvp' (VvpCatalog) implements org.apache.flink.table.catalog.CatalogTableProvider interface.
Hint: Please refer the document and use another catalog that supports schema evolution as the target catalog.
at org.apache.flink.table.sqlserver.utils.FormatValidatorExceptionUtils.newValidationException(FormatValidatorExceptionUtils.java:41)
at org.apache.flink.table.sqlserver.utils.ErrorConverter.formatException(ErrorConverter.java:123)
at org.apache.flink.table.sqlserver.utils.ErrorConverter.toErrorDetail(ErrorConverter.java:60)
at org.apache.flink.table.sqlserver.utils.ErrorConverter.toGrpcException(ErrorConverter.java:54)
at org.apache.flink.table.sqlserver.FlinkSqlServiceImpl.findCatalogObjectRefs(FlinkSqlServiceImpl.java:1105)
at org.apache.flink.table.sqlserver.proto.FlinkSqlServiceGrpc$MethodHandlers.invoke(FlinkSqlServiceGrpc.java:3761)
at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:172)
at io.grpc.PartialForwardingServerCallListener.onHalfClose(PartialForwardingServerCallListener.java:35)
at io.grpc.ForwardingServerCallListener.onHalfClose(ForwardingServerCallListener.java:23)
at io.grpc.ForwardingServerCallListener$SimpleForwardingServerCallListener.onHalfClose(ForwardingServerCallListener.java:40)
at org.apache.flink.table.sqlserver.interceptor.StatusInterceptor$1.onHalfClose(StatusInterceptor.java:116)
at io.grpc.PartialForwardingServerCallListener.onHalfClose(PartialForwardingServerCallListener.java:35)
at io.grpc.ForwardingServerCallListener.onHalfClose(ForwardingServerCallListener.java:23)
at io.grpc.ForwardingServerCallListener$SimpleForwardingServerCallListener.onHalfClose(ForwardingServerCallListener.java:40)
at org.apache.flink.table.sqlserver.interceptor.CancelHandlerRegister$1.onHalfClose(CancelHandlerRegister.java:59)
at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:331)
at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1HalfClosed.runInContext(ServerImpl.java:820)
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1147)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:622)
at java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.flink.table.api.ValidationException: CREATE TABLE ... AS TABLE ... statement requires target catalog 'vvp' (VvpCatalog) implements org.apache.flink.table.catalog.CatalogTableProvider interface.
Hint: Please refer the document and use another catalog that supports schema evolution as the target catalog.
at org.apache.flink.table.planner.operations.SqlCreateTableAsConverter.createUnresolvedCatalogTable(SqlCreateTableAsConverter.java:179)
at org.apache.flink.table.planner.operations.SqlUnresolvedConverter.inferUnresolvedTable(SqlUnresolvedConverter.java:159)
at org.apache.flink.table.planner.catalog.CatalogProber.inferUnresolvedSinkTable(CatalogProber.java:269)
at org.apache.flink.table.planner.calcite.ReferencesFinder$ReferencesVisitor.visitCreateTableAs(ReferencesFinder.java:335)
at org.apache.flink.table.planner.calcite.ReferencesFinder$ReferencesVisitor.visit(ReferencesFinder.java:178)
at org.apache.flink.table.planner.calcite.ReferencesFinder$ReferencesVisitor.visit(ReferencesFinder.java:127)
at org.apache.calcite.sql.SqlCall.accept(SqlCall.java:161)
at org.apache.calcite.sql.util.SqlBasicVisitor.visit(SqlBasicVisitor.java:54)
at org.apache.calcite.sql.SqlNodeList.accept(SqlNodeList.java:266)
at org.apache.flink.table.planner.calcite.ReferencesFinder.findReferences(ReferencesFinder.java:121)
at org.apache.flink.table.planner.delegation.ParserImpl.findCatalogObjectReferences(ParserImpl.java:151)
at org.apache.flink.table.sqlserver.execution.OperationExecutorImpl.findCatalogObjectRefs(OperationExecutorImpl.java:598)
at org.apache.flink.table.sqlserver.execution.DelegateOperationExecutor.lambda$findCatalogObjectRefs$13(DelegateOperationExecutor.java:183)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1729)
at org.apache.flink.table.sqlserver.context.SqlServerSecurityContext.runSecured(SqlServerSecurityContext.java:72)
at org.apache.flink.table.sqlserver.execution.DelegateOperationExecutor.wrapClassLoader(DelegateOperationExecutor.java:322)
at org.apache.flink.table.sqlserver.execution.DelegateOperationExecutor.lambda$wrapExecutor$36(DelegateOperationExecutor.java:349)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
... 3 more