content: /*
   * Copyright (c)  2022. Aliyun.com All right reserved. This software is the
   * confidential and proprietary information of Aliyun.com ("Confidential
   * Information"). You shall not disclose such Confidential Information and shall
   * use it only in accordance with the terms of the license agreement you entered
   * into with Aliyun.com.
   */
  
  package com.aliyun.fastmodel.transform.hologres.format;
  
  import java.util.ArrayList;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.stream.Collectors;
  
  import com.aliyun.fastmodel.common.utils.StripUtils;
  import com.aliyun.fastmodel.core.formatter.FastModelVisitor;
  import com.aliyun.fastmodel.core.tree.Comment;
  import com.aliyun.fastmodel.core.tree.Property;
  import com.aliyun.fastmodel.core.tree.QualifiedName;
  import com.aliyun.fastmodel.core.tree.datatype.BaseDataType;
  import com.aliyun.fastmodel.core.tree.datatype.DataTypeEnums;
  import com.aliyun.fastmodel.core.tree.datatype.GenericDataType;
  import com.aliyun.fastmodel.core.tree.datatype.IDataTypeName;
  import com.aliyun.fastmodel.core.tree.datatype.IDataTypeName.Dimension;
  import com.aliyun.fastmodel.core.tree.expr.BaseExpression;
  import com.aliyun.fastmodel.core.tree.expr.Identifier;
  import com.aliyun.fastmodel.core.tree.statement.CompositeStatement;
  import com.aliyun.fastmodel.core.tree.statement.table.AddCols;
  import com.aliyun.fastmodel.core.tree.statement.table.AddConstraint;
  import com.aliyun.fastmodel.core.tree.statement.table.AddPartitionCol;
  import com.aliyun.fastmodel.core.tree.statement.table.ChangeCol;
  import com.aliyun.fastmodel.core.tree.statement.table.ChangeCol.ChangeType;
  import com.aliyun.fastmodel.core.tree.statement.table.ColumnDefinition;
  import com.aliyun.fastmodel.core.tree.statement.table.CreateTable;
  import com.aliyun.fastmodel.core.tree.statement.table.DropCol;
  import com.aliyun.fastmodel.core.tree.statement.table.DropPartitionCol;
  import com.aliyun.fastmodel.core.tree.statement.table.DropTable;
  import com.aliyun.fastmodel.core.tree.statement.table.PartitionedBy;
  import com.aliyun.fastmodel.core.tree.statement.table.RenameTable;
  import com.aliyun.fastmodel.core.tree.statement.table.SetTableComment;
  import com.aliyun.fastmodel.core.tree.statement.table.SetTableProperties;
  import com.aliyun.fastmodel.core.tree.statement.table.UnSetTableProperties;
  import com.aliyun.fastmodel.core.tree.statement.table.constraint.BaseConstraint;
  import com.aliyun.fastmodel.core.tree.statement.table.constraint.PrimaryConstraint;
  import com.aliyun.fastmodel.core.tree.util.DataTypeUtil;
  import com.aliyun.fastmodel.core.tree.util.PropertyUtil;
  import com.aliyun.fastmodel.transform.api.util.StringJoinUtil;
  import com.aliyun.fastmodel.transform.hologres.client.property.HologresPropertyKey;
  import com.aliyun.fastmodel.transform.hologres.context.HologresTransformContext;
  import com.aliyun.fastmodel.transform.hologres.dialect.HologresVersion;
  import com.aliyun.fastmodel.transform.hologres.parser.tree.BeginWork;
  import com.aliyun.fastmodel.transform.hologres.parser.tree.CommitWork;
  import com.aliyun.fastmodel.transform.hologres.parser.tree.datatype.HologresDataTypeName;
  import com.aliyun.fastmodel.transform.hologres.parser.util.BuilderUtil;
  import com.aliyun.fastmodel.transform.hologres.parser.util.HologresPropertyUtil;
  import com.aliyun.fastmodel.transform.hologres.parser.visitor.HologresVisitor;
  import com.google.common.base.Joiner;
  import com.google.common.collect.Lists;
  import org.apache.commons.collections.CollectionUtils;
  import org.apache.commons.lang3.BooleanUtils;
  import org.apache.commons.lang3.StringUtils;
  
  import static java.util.stream.Collectors.joining;
  
  /**
   * 遍历node节点操作处理
   *
   * @author panguanjing
   * @date 2021/4/15
   */
  public class HologresAstVisitor extends FastModelVisitor implements HologresVisitor<Boolean, Integer> {
  
      /**
       * transform context
       */
      private final HologresTransformContext context;
  
      /**
       * hologres version
       */
      private final HologresVersion hologresVersion;
  
      /**
       * 用于修改动态表sql的前后缀
       */
      private static final String TASK_DEFINITION_STUB = "$_dataworks_system_$";
  
      public HologresAstVisitor(HologresTransformContext context, HologresVersion hologresVersion) {
          this.context = context;
          this.hologresVersion = hologresVersion;
      }
  
      public HologresAstVisitor(HologresTransformContext context) {
          this(context, HologresVersion.V1);
      }
  
      @Override
      public Boolean visitCompositeStatement(CompositeStatement compositeStatement, Integer context) {
          return super.visitCompositeStatement(compositeStatement, context);
      }
  
      @Override
      public Boolean visitCreateTable(CreateTable node, Integer indent) {
          Boolean foreignTable = isForeignTable(node);
          Boolean dynamicTable = isDynamicTable(node);
          if (foreignTable) {
              return visitCreateForeignTable(node, indent);
          } else if (dynamicTable) {
              return visitDynamicTable(node, indent);
          } else {
              return visitCreateInnerTable(node, indent);
          }
      }
  
      private Boolean visitDynamicTable(CreateTable node, Integer indent) {
          builder.append("BEGIN;\n");
          printExtensionIfNeed(node);
          builder.append("CREATE DYNAMIC TABLE ");
          if (node.isNotExists()) {
              builder.append("IF NOT EXISTS ");
          }
          String tableCode = getCode(node.getQualifiedName());
          builder.append(tableCode);
          String elementIndent = indentString(indent + 1);
          boolean columnEmpty = node.isColumnEmpty();
          if (!columnEmpty) {
              builder.append(" (\n");
              String column = formatColumnList(node.getColumnDefines(), elementIndent);
              builder.append(column);
              builder.append("\n").append(")");
          }
          if (!node.isPartitionEmpty()) {
              builder.append(" PARTITION BY LIST(").append(
                  node.getPartitionedBy().getColumnDefinitions().stream().map(x -> formatExpression(x.getColName()))
                      .collect(joining(","))).append(")");
          }
          //with
          List<Property> properties = node.getProperties();
          List<String> excludeKeys = Lists.newArrayList(
              HologresPropertyKey.DYNAMIC.getValue(),
              HologresPropertyKey.TASK_DEFINITION.getValue()
          );
          String p = formatProperty(elementIndent, properties.stream().filter(
              prop -> !excludeKeys.contains(prop.getName())
          ).collect(Collectors.toList()));
          if (StringUtils.isNotBlank(p)) {
              builder.append("\nWITH (\n");
              builder.append(p);
              builder.append("\n)");
          }
          String sql = PropertyUtil.getPropertyValue(properties, HologresPropertyKey.TASK_DEFINITION.getValue());
          //query
          if (StringUtils.isNotBlank(sql)) {
              builder.append(" AS\n");
              builder.append(sql);
              if (!sql.endsWith(";")) {
                  builder.append(";");
              }
          }
          builder.append("\n");
          builder.append("COMMIT;");
          if (node.getCommentValue() != null) {
              builder.append("\n");
              BuilderUtil.addTransaction(builder, () -> commentTable(tableCode, node.getCommentValue()));
          }
          return true;
      }
  
      private String formatProperty(String indent, List<Property> properties) {
          if (properties == null || properties.isEmpty()) {
              return StringUtils.EMPTY;
          }
          return properties.stream().map(
              x -> indent + x.getName() + "=" + formatStringLiteral(x.getValue())
          ).collect(joining(",\n"));
      }
  
      private Boolean isDynamicTable(CreateTable node) {
          if (node == null) {
              return false;
          }
          List<Property> properties = node.getProperties();
          if (CollectionUtils.isEmpty(properties)) {
              return false;
          }
          return properties.stream().anyMatch(p -> StringUtils.equalsIgnoreCase(p.getName(), HologresPropertyKey.DYNAMIC.getValue()));
      }
  
      private Boolean visitCreateInnerTable(CreateTable node, Integer indent) {
          boolean columnEmpty = node.isColumnEmpty();
          boolean executable = !columnEmpty;
          builder.append("BEGIN;\n");
          printExtensionIfNeed(node);
          builder.append("CREATE TABLE ");
          if (node.isNotExists()) {
              builder.append("IF NOT EXISTS ");
          }
          String tableCode = getCode(node.getQualifiedName());
          builder.append(tableCode);
          String elementIndent = indentString(indent + 1);
          List<ColumnDefinition> columnDefines = merge(node.getColumnDefines(), node.getPartitionedBy());
          if (!columnEmpty) {
              builder.append(" (\n");
              String columnList = formatColumnList(columnDefines, elementIndent);
              builder.append(columnList);
              if (!node.isConstraintEmpty()) {
                  appendConstraint(node, indent);
              }
              builder.append("\n").append(")");
          }
          if (!node.isPartitionEmpty()) {
              builder.append(" PARTITION BY LIST(").append(
                  node.getPartitionedBy().getColumnDefinitions().stream().map(x -> formatExpression(x.getColName()))
                      .collect(joining(","))).append(")");
          }
          builder.append(";\n");
          List<Property> properties = node.getProperties();
          if (CollectionUtils.isNotEmpty(properties)) {
              String propertiesValue = buildSetProperties(node.getQualifiedName(), properties);
              builder.append(propertiesValue);
          }
          if (node.getComment() != null && node.getComment().getComment() != null) {
              builder.append(commentTable(tableCode, node.getCommentValue()));
          }
          if (!columnEmpty) {
              for (ColumnDefinition columnDefinition : columnDefines) {
                  if (columnDefinition.getComment() == null || columnDefinition.getComment().getComment() == null) {
                      continue;
                  }
                  builder.append("\n");
                  builder.append(commentColumn(tableCode, formatColName(columnDefinition.getColName(), 0), columnDefinition.getCommentValue()));
              }
          }
          builder.append("\n");
          builder.append("COMMIT;");
          return executable;
      }
  
      private void printExtensionIfNeed(CreateTable node) {
          if (node.isColumnEmpty()) {
              return;
          }
          Optional<ColumnDefinition> first = node.getColumnDefines().stream()
              .filter(c -> c.getDataType() != null)
              .filter(c -> StringUtils.equalsIgnoreCase(c.getDataType().getTypeName().getValue(), HologresDataTypeName.ROARING_BITMAP.getValue()))
              .findFirst();
          if (!first.isPresent()) {
              return;
          }
          builder.append("CREATE EXTENSION IF NOT EXISTS ROARINGBITMAP;\n");
      }
  
      private Boolean visitCreateForeignTable(CreateTable node, Integer indent) {
          boolean columnEmpty = node.isColumnEmpty();
          boolean executable = !columnEmpty;
          builder.append("BEGIN;\n");
          builder.append("CREATE FOREIGN TABLE ");
          if (node.isNotExists()) {
              builder.append("IF NOT EXISTS ");
          }
          String tableCode = getCode(node.getQualifiedName());
          builder.append(tableCode);
          String elementIndent = indentString(indent + 1);
          List<ColumnDefinition> columnDefines = merge(node.getColumnDefines(), node.getPartitionedBy());
          if (!columnEmpty) {
              builder.append(" (\n");
              String columnList = formatColumnList(columnDefines, elementIndent);
              builder.append(columnList);
              if (!node.isConstraintEmpty()) {
                  appendConstraint(node, indent);
              }
              builder.append("\n").append(")");
          }
          String serverAndOptions = buildServerAndOptions(node);
          builder.append(serverAndOptions);
          builder.append(";\n");
          builder.append("COMMIT;");
          return executable;
      }
  
      private List<ColumnDefinition> merge(List<ColumnDefinition> columnDefines, PartitionedBy partitionedBy) {
          if (partitionedBy == null || !partitionedBy.isNotEmpty()) {
              return columnDefines;
          }
          List<ColumnDefinition> list = Lists.newArrayList(columnDefines);
          List<ColumnDefinition> partitionedByColumnDefinitions = partitionedBy.getColumnDefinitions();
          for (ColumnDefinition columnDefinition : partitionedByColumnDefinitions) {
              if (contains(list, columnDefinition)) {
                  continue;
              }
              list.add(columnDefinition);
          }
          return list;
      }
  
      private boolean contains(List<ColumnDefinition> list, ColumnDefinition columnDefinition) {
          return list.stream().anyMatch(definition -> Objects.equals(definition.getColName(), columnDefinition.getColName()));
      }
  
      private void appendConstraint(CreateTable node, Integer indent) {
          for (BaseConstraint next : node.getConstraintStatements()) {
              //hologres只有primary key定义
              if (!(next instanceof PrimaryConstraint)) {
                  continue;
              }
              builder.append(",\n");
              process(next, indent + 1);
          }
      }
  
      @Override
      public Boolean visitPrimaryConstraint(PrimaryConstraint primaryConstraint, Integer indent) {
          builder.append(indentString(indent)).append("PRIMARY KEY(");
          builder.append(
              primaryConstraint.getColNames().stream().map(
                  c -> formatColName(c, 0)
              ).collect(joining(",")));
          builder.append(")");
          return true;
      }
  
      private String callSetProperty(String code, String key, String value) {
          String result = HologresPropertyUtil.getPropertyValue(hologresVersion, key, value);
          //将code中的双引号去除
          String strip = StripUtils.removeDoubleStrip(code);
          //如果是列属性，那么
          String format = "CALL SET_TABLE_PROPERTY('%s', '%s', '%s');";
          return String.format(format, strip, key, result);
      }
  
      private String commentTable(String code, String comment) {
          String format = "COMMENT ON TABLE %s IS %s;";
          if (comment == null) {
              return String.format(format, code, "NULL");
          }
          return String.format(format, code, formatStringLiteral(comment));
      }
  
      private String commentColumn(String code, String column, String comment) {
          String format = "COMMENT ON COLUMN %s.%s IS %s;";
          if (comment == null) {
              return String.format(format, code, column, "NULL");
          }
          return String.format(format, code, column, formatStringLiteral(comment));
      }
  
      @Override
      protected String formatColumnDefinition(ColumnDefinition column, Integer max) {
          StringBuilder sb = appendNameAndType(column, max);
          if (column.getDataType() == null) {
              //如果没有数据类型，那么只返回名称即可
              return sb.toString();
          }
          boolean isPrimary = column.getPrimary() != null && column.getPrimary();
          if (isPrimary) {
              sb.append(" PRIMARY KEY");
          }
          boolean isNotNull = column.getNotNull() != null && column.getNotNull();
          if (!isPrimary && isNotNull) {
              sb.append(" NOT NULL");
          }
          if (column.getDefaultValue() != null) {
              sb.append(" DEFAULT ").append(formatExpression(column.getDefaultValue()));
          }
          return sb.toString();
      }
  
      private StringBuilder appendNameAndType(ColumnDefinition column, Integer max) {
          BaseDataType dataType = column.getDataType();
          BaseDataType convert = convert(dataType);
          String colName = formatColName(column.getColName(), max);
          if (convert == null) {
              return new StringBuilder().append(colName);
          }
          String expression = formatExpression(convert);
          return new StringBuilder()
              .append(colName)
              .append(" ")
              .append(expression);
      }
  
      @Override
      protected String formatColName(Identifier colName, Integer size) {
          String value = new HologresExpressionVisitor(context).visitIdentifier(colName, null);
          return StringUtils.rightPad(value, size);
      }
  
      @Override
      protected BaseDataType convert(BaseDataType dataType) {
          if (dataType == null) {
              return null;
          }
          IDataTypeName typeName = dataType.getTypeName();
          if (StringUtils.equalsIgnoreCase(typeName.getValue(), DataTypeEnums.STRING.getValue())) {
              return new GenericDataType(DataTypeEnums.TEXT.name());
          } else if (StringUtils.equalsIgnoreCase(typeName.getValue(), DataTypeEnums.DATETIME.getValue())) {
              return DataTypeUtil.simpleType(DataTypeEnums.TIMESTAMP);
          } else if (typeName.getDimension() == Dimension.MULTIPLE) {
              return DataTypeUtil.simpleType(DataTypeEnums.JSON);
          }
          return dataType;
      }
  
      @Override
      public Boolean visitAddCols(AddCols addCols, Integer context) {
          BuilderUtil.addTransaction(builder, () -> {
              StringBuilder stringBuilder = new StringBuilder();
              stringBuilder.append("ALTER TABLE IF EXISTS ").append(getCode(addCols.getQualifiedName()));
              String columnList = addCols.getColumnDefineList().stream()
                  .map(element -> " ADD COLUMN " + appendNameAndType(element, 0)).collect(joining(","));
              stringBuilder.append(columnList).append(";");
              for (ColumnDefinition columnDefinition : addCols.getColumnDefineList()) {
                  if (columnDefinition.getCommentValue() != null) {
                      stringBuilder.append("\n");
                      stringBuilder.append(
                          commentColumn(getCode(addCols.getQualifiedName()),
                              formatColName(columnDefinition.getColName(), 0),
                              columnDefinition.getCommentValue()));
                  }
              }
              return stringBuilder.toString();
          });
          return true;
      }
  
      @Override
      public Boolean visitDropTable(DropTable dropTable, Integer context) {
          builder.append("DROP TABLE IF EXISTS ").append(getCode(dropTable.getQualifiedName()));
          return true;
      }
  
      @Override
      public Boolean visitSetTableComment(SetTableComment setTableComment, Integer context) {
          BuilderUtil.addTransaction(builder, () -> {
              return commentTable(getCode(setTableComment.getQualifiedName()), setTableComment.getComment().getComment());
          });
          return true;
      }
  
      @Override
      public Boolean visitSetTableProperties(SetTableProperties setTableProperties, Integer indent) {
          List<Property> propertyList = setTableProperties.getProperties().stream().filter(
              p -> {
                  HologresPropertyKey byValue = HologresPropertyKey.getByValue(p.getName());
                  return byValue != null && byValue.isSupportPrint();
              }
          ).collect(Collectors.toList());
          if (propertyList.isEmpty()) {
              return false;
          }
          boolean useAlterTableSetSentence = this.context.isUseAlterTableSetSentence();
          if (useAlterTableSetSentence) {
              BuilderUtil.addTransaction(builder, () -> buildSetPropertiesUseAlter(setTableProperties, propertyList));
          } else {
              BuilderUtil.addTransaction(builder, () -> buildSetProperties(setTableProperties.getQualifiedName(), propertyList));
          }
          return true;
      }
  
      private String buildSetPropertiesUseAlter(SetTableProperties setTableProperties, List<Property> propertyList) {
          //ALTER TABLE <schema_name>.<table_name> SET (dictionary_encoding_columns = '[columnName{:[on|off|auto]}[,...]]');
          //针对Task_definition进行单独设置
          String prefix = "ALTER TABLE " + getCode(setTableProperties.getQualifiedName())
              + " SET ";
          String propertyValue = PropertyUtil.getPropertyValue(propertyList, HologresPropertyKey.TASK_DEFINITION.getValue());
          List<String> list = Lists.newArrayList();
          if (propertyValue != null) {
              String taskDefinitionBuilder = prefix + HologresPropertyKey.TASK_DEFINITION.getValue() + " = "
                  + TASK_DEFINITION_STUB
                  + StringUtils.LF
                  + propertyValue
                  + StringUtils.LF
                  + TASK_DEFINITION_STUB + ";";
              list.add(taskDefinitionBuilder);
          }
          String value = propertyList.stream().filter(p -> {
              return !StringUtils.equalsIgnoreCase(p.getName(), HologresPropertyKey.TASK_DEFINITION.getValue());
          }).map(p -> {
              String result = HologresPropertyUtil.getPropertyValue(hologresVersion, p.getName(), p.getValue());
              //将code中的双引号去除
              return p.getName() + "=" + StripUtils.addStrip(result);
          }).collect(joining(",", "(", ")"));
          list.add(prefix + value + ";");
          return String.join("\n", list);
      }
  
      private String buildSetProperties(QualifiedName qualifiedName, List<Property> properties) {
          StringBuilder stringBuilder = new StringBuilder();
          Iterator<Property> iterator = properties.iterator();
          if (iterator.hasNext()) {
              Property p = iterator.next();
              String value = callSetProperty(getCode(qualifiedName), p.getName(), p.getValue());
              stringBuilder.append(value);
              while (iterator.hasNext()) {
                  p = iterator.next();
                  String str = callSetProperty(getCode(qualifiedName), p.getName(), p.getValue());
                  if (StringUtils.isBlank(str)) {
                      continue;
                  }
                  stringBuilder.append("\n");
                  stringBuilder.append(str);
              }
          }
          return stringBuilder.toString();
      }
  
      @Override
      public Boolean visitDropCol(DropCol dropCol, Integer context) {
          super.visitDropCol(dropCol, context);
          return false;
      }
  
      @Override
      public Boolean visitAddPartitionCol(AddPartitionCol addPartitionCol, Integer context) {
          super.visitAddPartitionCol(addPartitionCol, context);
          return false;
      }
  
      @Override
      public Boolean visitDropPartitionCol(DropPartitionCol dropPartitionCol, Integer context) {
          super.visitDropPartitionCol(dropPartitionCol, context);
          return false;
      }
  
      @Override
      public Boolean visitUnSetTableProperties(UnSetTableProperties unSetTableProperties, Integer context) {
          super.visitUnSetTableProperties(unSetTableProperties, context);
          return false;
      }
  
      @Override
      public Boolean visitAddConstraint(AddConstraint addConstraint, Integer context) {
          super.visitAddConstraint(addConstraint, context);
          return false;
      }
  
      @Override
      public Boolean visitRenameTable(RenameTable renameTable, Integer context) {
          builder.append("ALTER TABLE ");
          builder.append(getCode(renameTable.getQualifiedName()));
          builder.append(" RENAME TO ").append(renameTable.getTarget().getSuffix());
          return true;
      }
  
      @Override
      public Boolean visitChangeCol(ChangeCol changeCol, Integer context) {
          Identifier oldColName = changeCol.getOldColName();
          Identifier newColName = changeCol.getNewColName();
          String code = getCode(changeCol.getQualifiedName());
          List<String> changeValue = Lists.newArrayList();
          if (Objects.equals(oldColName, newColName)) {
              BaseExpression defaultValue = changeCol.getDefaultValue();
              if (defaultValue != null) {
                  String builder = "ALTER TABLE " + code
                      + " ALTER COLUMN " + formatColName(oldColName, 0)
                      + " SET DEFAULT " + formatExpression(defaultValue)
                      + ";";
                  changeValue.add(builder);
              }
              //如果改了默认值
              if (changeCol.change(ChangeType.DEFAULT_VALUE)) {
                  if (defaultValue == null) {
                      String builder = "ALTER TABLE " + code
                          + " ALTER COLUMN " + formatColName(oldColName, 0)
                          + " DROP DEFAULT;";
                      changeValue.add(builder);
                  }
              }
              Comment comment = changeCol.getColumnDefinition().getComment();
              if (comment != null) {
                  StringBuilder builder = new StringBuilder();
                  String commentColumn = commentColumn(code, formatColName(newColName, 0), comment.getComment());
                  builder.append(commentColumn);
                  changeValue.add(builder.toString());
              }
          } else {
              String builder = "ALTER TABLE " + code
                  + " RENAME COLUMN " + formatExpression(oldColName)
                  + " TO " + formatExpression(newColName)
                  + ";";
              changeValue.add(builder);
          }
          if (changeValue.isEmpty()) {
              super.visitChangeCol(changeCol, context);
              return false;
          }
          String join = Joiner.on("\n").join(changeValue);
          BuilderUtil.addTransaction(builder, () -> join);
          return true;
      }
  
      @Override
      protected String formatExpression(BaseExpression baseExpression) {
          return new HologresExpressionVisitor(context).process(baseExpression);
      }
  
      @Override
      protected String getCode(QualifiedName qualifiedName) {
          //hologres的2.x版本不支持3段式的创建，1.x支持，为了兼容统一采用2段式的创建
          QualifiedName tableName = StringJoinUtil.join(
              null,
              this.context.getSchema(),
              qualifiedName.getSuffix()
          );
          return formatName(tableName);
      }
  
      @Override
      public Boolean visitBeginWork(BeginWork beginWork, Integer context) {
          builder.append("BEGIN;");
          return true;
      }
  
      @Override
      public Boolean visitCommitWork(CommitWork commitWork, Integer context) {
          builder.append("COMMIT;");
          return null;
      }
  
      private Boolean isForeignTable(CreateTable node) {
          if (CollectionUtils.isEmpty(node.getProperties())) {
              return false;
          }
          Optional<Property> foreignTablePropertyOpt = node.getProperties().stream().filter(property ->
              StringUtils.equalsIgnoreCase(HologresPropertyKey.FOREIGN.getValue(), property.getName())
                  && BooleanUtils.toBoolean(property.getValue())).findAny();
          return foreignTablePropertyOpt.isPresent();
      }
  
      private String buildServerAndOptions(CreateTable node) {
          StringBuilder sb = new StringBuilder();
          Optional<Property> serverOpt = node.getProperties().stream().filter(property ->
              StringUtils.equalsIgnoreCase(HologresPropertyKey.SERVER_NAME.getValue(), property.getName())).findAny();
          serverOpt.ifPresent(server -> sb.append("\nSERVER ").append(server.getValue()).append("\n"));
  
          List<String> options = new ArrayList<>();
          Optional<Property> projectNameOpt = node.getProperties().stream().filter(property ->
              StringUtils.equalsIgnoreCase("project_name", property.getName())).findAny();
          projectNameOpt.ifPresent(projectName -> options.add("project_name " + StripUtils.addStrip(projectName.getValue())));
          Optional<Property> tableNameOpt = node.getProperties().stream().filter(property ->
              StringUtils.equalsIgnoreCase("table_name", property.getName())).findAny();
          tableNameOpt.ifPresent(tableName -> options.add("table_name " + StripUtils.addStrip(tableName.getValue())));
          if (CollectionUtils.isNotEmpty(options)) {
              sb.append("OPTIONS(").append(StringUtils.join(options, ", ")).append(")");
          }
          return sb.toString();
      }
  }
  