presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java/StatementAnalyzer/visitCreateTableAsSelect
class StatementAnalyzer:        @Override
        protected Scope visitCreateTableAsSelect(CreateTableAsSelect node, Optional<Scope> scope)
        {
            analysis.setUpdateType("CREATE TABLE");

            // turn this into a query that has a new table writer node on top.
            QualifiedObjectName targetTable = createQualifiedObjectName(session, node, node.getName());

            Optional<TableHandle> targetTableHandle = metadata.getTableHandle(session, targetTable);
            if (targetTableHandle.isPresent()) {
                if (node.isNotExists()) {
                    analysis.setCreate(new Analysis.Create(
                            Optional.of(targetTable),
                            Optional.empty(),
                            Optional.empty(),
                            node.isWithData(),
                            true));
                    return createAndAssignScope(node, scope, Field.newUnqualified("rows", BIGINT));
                }
                throw semanticException(TABLE_ALREADY_EXISTS, node, "Destination table '%s' already exists", targetTable);
            }

            validateProperties(node.getProperties(), scope);

            accessControl.checkCanCreateTable(session.toSecurityContext(), targetTable);

            // analyze the query that creates the table
            Scope queryScope = analyze(node.getQuery(), createScope(scope));

            ImmutableList.Builder<ColumnMetadata> columns = ImmutableList.builder();

            // analyze target table columns and column aliases
            if (node.getColumnAliases().isPresent()) {
                validateColumnAliases(node.getColumnAliases().get(), queryScope.getRelationType().getVisibleFieldCount());

                int aliasPosition = 0;
                for (Field field : queryScope.getRelationType().getVisibleFields()) {
                    if (field.getType().equals(UNKNOWN)) {
                        throw semanticException(COLUMN_TYPE_UNKNOWN, node, "Column type is unknown at position %s", queryScope.getRelationType().indexOf(field) + 1);
                    }
                    columns.add(new ColumnMetadata(node.getColumnAliases().get().get(aliasPosition).getValue(), field.getType()));
                    aliasPosition++;
                }
            }
            else {
                validateColumns(node, queryScope.getRelationType());
                columns.addAll(queryScope.getRelationType().getVisibleFields().stream()
                        .map(field -> new ColumnMetadata(field.getName().get(), field.getType()))
                        .collect(toImmutableList()));
            }

            // create target table metadata
            CatalogName catalogName = metadata.getCatalogHandle(session, targetTable.getCatalogName())
                    .orElseThrow(() -> new PrestoException(NOT_FOUND, "Catalog does not exist: " + targetTable.getCatalogName()));

            Map<String, Object> properties = metadata.getTablePropertyManager().getProperties(
                    catalogName,
                    targetTable.getCatalogName(),
                    mapFromProperties(node.getProperties()),
                    session,
                    metadata,
                    analysis.getParameters());

            ConnectorTableMetadata tableMetadata = new ConnectorTableMetadata(targetTable.asSchemaTableName(), columns.build(), properties, node.getComment());

            // analyze target table layout
            Optional<NewTableLayout> newTableLayout = metadata.getNewTableLayout(session, targetTable.getCatalogName(), tableMetadata);

            Set<String> columnNames = columns.build().stream()
                    .map(ColumnMetadata::getName)
                    .collect(toImmutableSet());

            newTableLayout.ifPresent(layout -> {
                if (!columnNames.containsAll(layout.getPartitionColumns())) {
                    throw new PrestoException(NOT_SUPPORTED, "INSERT must write all distribution columns: " + layout.getPartitionColumns());
                }
            });

            analysis.setCreate(new Analysis.Create(
                    Optional.of(targetTable),
                    Optional.of(tableMetadata),
                    newTableLayout,
                    node.isWithData(),
                    false));

            return createAndAssignScope(node, scope, Field.newUnqualified("rows", BIGINT));
        }