Ensure that the column mappings for the shared fields between the old and
new schema haven't changed.

@param oldSchema
@param newSchema
@return true if the mappings are compatible, false if not.
326903
private static boolean mappingCompatible(EntitySchema oldSchema,
      EntitySchema newSchema) {
    for (FieldMapping oldFieldMapping : oldSchema.getColumnMappingDescriptor()
        .getFieldMappings()) {
      FieldMapping newFieldMapping = newSchema.getColumnMappingDescriptor()
          .getFieldMapping(oldFieldMapping.getFieldName());
      if (newFieldMapping != null) {
        if (!oldFieldMapping.equals(newFieldMapping)) {
          return false;
        }
      }
    }
    return true;
  }
873615
public void setSchemaFacet(java.util.Collection<SchemaFacet> schemaFacet) {
        if (schemaFacet == null) {
            this.schemaFacet = null;
            return;
        }

        this.schemaFacet = new java.util.ArrayList<SchemaFacet>(schemaFacet);
    }
439399
void compareTypes(Schema oldSchema,
                                           Schema newSchema,
                                           List<Message> messages,
                                           String name) {
        oldSchema = stripOptionalTypeUnion(oldSchema);
        newSchema = stripOptionalTypeUnion(newSchema);
        Schema.Type oldType = oldSchema.getType();
        Schema.Type newType = newSchema.getType();

        if(oldType != Type.UNION && newType == Type.UNION) {
            boolean compatibleTypeFound = false;
            for(Schema s: newSchema.getTypes()) {
                if((oldType != Type.RECORD && oldType == s.getType())
                   || (oldType == Type.RECORD && s.getType() == Type.RECORD && oldSchema.getName()
                                                                                        .equals(s.getName()))) {
                    compareTypes(oldSchema, s, messages, name);
                    compatibleTypeFound = true;
                    break;
                }
            }

            if(compatibleTypeFound) {
                messages.add(new Message(Level.INFO,
                                         "Type change from " + oldType + " to " + newType
                                                 + " for field " + name
                                                 + ". The new union includes the original type."));
            } else {
                messages.add(new Message(Level.ERROR,
                                         "Incompatible type change from "
                                                 + oldType
                                                 + " to "
                                                 + newType
                                                 + " for field "
                                                 + name
                                                 + ". The new union does not include the original type."));
            }
        } else if(oldType == Type.RECORD) {
            if(!_recordStack.contains(oldSchema.getName())) {
                _recordStack.add(oldSchema.getName());
                compareRecordTypes(oldSchema, newSchema, messages, name);
                _recordStack.remove(oldSchema.getName());
            }
        } else if(oldType == Type.ENUM) {
            compareEnumTypes(oldSchema, newSchema, messages, name);
        } else if(oldType == Type.ARRAY) {
            compareArrayTypes(oldSchema, newSchema, messages, name);
        } else if(oldType == Type.MAP) {
            compareMapTypes(oldSchema, newSchema, messages, name);
        } else if(oldType == Type.UNION) {
            compareUnionTypes(oldSchema, newSchema, messages, name);
        } else if(oldType == Type.FIXED) {
            compareFixedTypes(oldSchema, newSchema, messages, name);
        } else {
            comparePrimitiveTypes(oldSchema, newSchema, messages, name);
        }
    }
196893
public void setKeySchema(java.util.Collection<KeySchemaElement> keySchema) {
        if (keySchema == null) {
            this.keySchema = null;
            return;
        }

        this.keySchema = new java.util.ArrayList<KeySchemaElement>(keySchema);
    }
1059567
public void updateSchemaVersion(VersionComparator version) throws SQLException {
        PreparedStatement statement = null;
        PreparedStatement updateStatement = null;
        ResultSet results = null;
        try {
            statement = connection.prepareStatement("SELECT \"VERSION\" FROM \"SCHEMAVERSION\"");
            results = statement.executeQuery();

            if (results.next()) {
                final VersionComparator currentVersion = new VersionComparator(results.getString(1));
                if (version == null || currentVersion.isNewerThan(version)) {
                    return;
                }
                updateStatement = connection.prepareStatement("UPDATE \"SCHEMAVERSION\" SET \"VERSION\" = ?");
            } else {
                // Does not exist. Populate schema table with current running version
                version = new VersionComparator(Config.getInstance().getApplicationVersion());
                updateStatement = connection.prepareStatement("INSERT INTO \"SCHEMAVERSION\" (\"VERSION\") VALUES (?)");
            }

            LOGGER.debug("Updating database schema to: " + version.toString());

            updateStatement.setString(1, version.toString());
            updateStatement.executeUpdate();
            connection.commit();
        } finally {
            DbUtil.close(results);
            DbUtil.close(updateStatement);
            DbUtil.close(statement);
            //DbUtil.close(connection); // do not close connection
        }
    }
380630
public void setSchemas(java.util.Collection<String> schemas) {
        if (schemas == null) {
            this.schemas = null;
            return;
        }

        this.schemas = new java.util.ArrayList<String>(schemas);
    }
1364659
public static Schema subSetOf(String newName, Schema schema, String... subSetFields) {
		List<Field> newSchema = new ArrayList<Field>();
		for(String subSetField: subSetFields) {
			newSchema.add(schema.getField(subSetField));
		}
		return new Schema(newName, newSchema);
	}
428018
private static boolean newPojoHasNewOrRemovedFields(
			LinkedOptionalMap<Field, TypeSerializerSnapshot<?>> fieldSerializerSnapshots,
			PojoSerializer<?> newPojoSerializer) {
		int numRemovedFields = fieldSerializerSnapshots.absentKeysOrValues().size();
		int numPreexistingFields = fieldSerializerSnapshots.size() - numRemovedFields;

		boolean hasRemovedFields = numRemovedFields > 0;
		boolean hasNewFields = newPojoSerializer.getFields().length - numPreexistingFields > 0;
		return hasRemovedFields || hasNewFields;
	}
444409
private boolean validateCompatibleWithTableColumns(
      EntitySchema entitySchema1, EntitySchema entitySchema2) {

    // Populate two collections of field mappings. One that contains all
    // of the column mappings, and one that contains the keyAsColumn
    // mappings from the first schema. These will be used to compare
    // against the second schema.
    Set<String> entitySchema1Columns = new HashSet<String>();
    List<String> entitySchema1KeyAsColumns = new ArrayList<String>();
    for (FieldMapping fieldMapping1 : entitySchema1
        .getColumnMappingDescriptor().getFieldMappings()) {
      if (fieldMapping1.getMappingType() == MappingType.COLUMN ||
          fieldMapping1.getMappingType() == MappingType.COUNTER) {
        entitySchema1Columns.add(getColumnValue(fieldMapping1));
      } else if (fieldMapping1.getMappingType() == MappingType.KEY_AS_COLUMN) {
        entitySchema1KeyAsColumns.add(getColumnValue(fieldMapping1));
      }
    }

    // For each field mapping in the second entity schema, we want to
    // validate the following:
    //
    // 1. That each column mapping in it doesn't map to the same column
    // as a column mapping in the first schema.
    //
    // 2. That each column mapping in it doesn't "startsWith()" with a
    // keyAsColumn mapping in the first schema, where the keyAsColumn
    // mapping value is "columnfamily:prefix".
    //
    // 3. That each keyAsColumn mapping in it isn't a prefix of one of
    // the first schema's column mappings.
    //
    // 4. That each keyAsColumn mapping in it isn't a prefix of one fo
    // the first schema's keyAsColumn mappings, and one of the first
    // schema's mappings isn't a prefix of this schema's keyAsColumn
    // mappings.
    for (FieldMapping fieldMapping2 : entitySchema2
        .getColumnMappingDescriptor().getFieldMappings()) {
      if (fieldMapping2.getMappingType() == MappingType.COLUMN ||
          fieldMapping2.getMappingType() == MappingType.COUNTER) {
        String value = getColumnValue(fieldMapping2);
        if (entitySchema1Columns.contains(value)) {
          LOG.warn("Field: " + fieldMapping2.getFieldName()
              + " has a table column conflict with a column mapped field in "
              + entitySchema1.getName());
          return false;
        }
        for (String keyAsColumn : entitySchema1KeyAsColumns) {
          if (value.startsWith(keyAsColumn)) {
            LOG.warn("Field: "
                + fieldMapping2.getFieldName()
                + " has a table column conflict with a keyAsColumn mapped field in "
                + entitySchema1.getName());
            return false;
          }
        }
      } else if (fieldMapping2.getMappingType() == MappingType.KEY_AS_COLUMN) {
        String entitySchema2KeyAsColumn = getColumnValue(fieldMapping2);
        for (String entitySchema1KeyAsColumn : entitySchema1KeyAsColumns) {
          if (entitySchema1KeyAsColumn.startsWith(entitySchema2KeyAsColumn)) {
            LOG.warn("Field "
                + fieldMapping2.getFieldName()
                + " has a table keyAsColumn conflict with a keyAsColumn mapped field in "
                + entitySchema1.getName());
            return false;
          }
        }
        for (String entitySchema1Column : entitySchema1Columns) {
          if (entitySchema1Column.startsWith(entitySchema2KeyAsColumn)) {
            LOG.warn("Field "
                + fieldMapping2.getFieldName()
                + " has a table keyAsColumn conflict with a column mapped field in "
                + entitySchema1.getName());
            return false;
          }
        }
      }
    }
    return true;
  }
515448
private void recordCompatibilityInfo(JMFType access, JSchema accSchema, JMFType encoding,
      JSchema encSchema) throws JMFSchemaViolationException
  {
    if (access instanceof JSPrimitive)
      if (encoding instanceof JSPrimitive)
        recordCompatibilityInfo((JSPrimitive)access, accSchema, (JSPrimitive)encoding,
          encSchema);
      else if (encoding instanceof JSVariant)
        checkForDeletingVariant(access, (JSVariant)encoding, false, accSchema, encSchema);
      else {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) JmfTr.debug(this, tc, "Compatibility violation (type mismatch) for JSPrimitive");
        violation(access, encoding);
       }
    else if (access instanceof JSEnum)
      if (encoding instanceof JSEnum)
        recordCompatibilityInfo((JSEnum)access, accSchema, (JSEnum)encoding, encSchema);
      else if (encoding instanceof JSVariant)
        checkForDeletingVariant(access, (JSVariant)encoding, false, accSchema, encSchema);
      else {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) JmfTr.debug(this, tc, "Compatibility violation (type mismatch) for JSEnum");
        violation(access, encoding);
      }
    else if (access instanceof JSDynamic)
      if (encoding instanceof JSDynamic)
        recordOnePair((JSField)access, accSchema, (JSField)encoding, encSchema);
      else if (encoding instanceof JSVariant)
        checkForDeletingVariant(access, (JSVariant)encoding, false, accSchema, encSchema);
      else {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) JmfTr.debug(this, tc, "Compatibility violation (type mismatch) for JSDynamic");
        violation(access, encoding);
      }
    else if (access instanceof JSRepeated)
      if (encoding instanceof JSRepeated)
        recordCompatibilityInfo(((JSRepeated)access).getItemType(), accSchema,
          ((JSRepeated)encoding).getItemType(), encSchema);
      else if (encoding instanceof JSVariant)
        checkForDeletingVariant(access, (JSVariant)encoding, false, accSchema, encSchema);
      else {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) JmfTr.debug(this, tc, "Compatibility violation (type mismatch) for JSRepeated");
        violation(access, encoding);
      }
    else if (access instanceof JSTuple)
      if (encoding instanceof JSTuple)
        recordCompatibilityInfo((JSTuple)access, accSchema, (JSTuple)encoding, encSchema);
      else if (encoding instanceof JSVariant)
        checkForDeletingVariant(access, (JSVariant)encoding, false, accSchema, encSchema);
      else {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) JmfTr.debug(this, tc, "Compatibility violation (type mismatch) for JSTuple");
        violation(access, encoding);
      }
    else if (access instanceof JSVariant) {
      JSVariant accVar = (JSVariant)access;
      if (encoding instanceof JSVariant) {
        JSVariant encVar = (JSVariant)encoding;
        recordCompatibilityInfo(accVar, accSchema, encVar, encSchema);
      } else
        checkForDeletingVariant(encoding, accVar, true, accSchema, encSchema);
    }
  }
1242511
private void checkUpdateKeyColumns(Map<String, EventColumn> oldKeyColumns, Map<String, EventColumn> keyColumns) {
        // 在变更前没有主键的情况
        if (oldKeyColumns.size() == 0) {
            return;
        }
        // 变更后的主键数据大于变更前的，不符合
        if (keyColumns.size() > oldKeyColumns.size()) {
            return;
        }
        // 主键没有变更，把所有变更前的主键拷贝到变更后的主键中.
        if (keyColumns.size() == 0) {
            keyColumns.putAll(oldKeyColumns);
            return;
        }

        // 把old中存在而new中不存在的主键拷贝到new中
        if (oldKeyColumns.size() != keyColumns.size()) {
            for (String oldKey : oldKeyColumns.keySet()) {
                if (keyColumns.get(oldKey) == null) {
                    keyColumns.put(oldKey, oldKeyColumns.get(oldKey));
                }
            }
        }
    }
899422
void compareArrayTypes(Schema oldSchema,
                                                Schema newSchema,
                                                List<Message> messages,
                                                String name) {
        if(oldSchema == null || newSchema == null || oldSchema.getType() != Schema.Type.ARRAY) {
            throw new IllegalArgumentException("Old schema must be ARRAY type. Name=" + name
                                               + ". Type=" + oldSchema);
        }

        if(newSchema.getType() != Schema.Type.ARRAY) {
            messages.add(new Message(Level.ERROR, "Illegal type change from " + oldSchema.getType()
                                                  + " to " + newSchema.getType() + " for field "
                                                  + name));
            return;
        }

        // Compare the array element types
        compareTypes(oldSchema.getElementType(),
                     newSchema.getElementType(),
                     messages,
                     name + ".<array element>");
    }
1260297
public function isVersioningService()
    {
        $schema = $this->utils->getCurrentSchema();
        if (isset($schema->{'x-versioning'}) && $schema->{'x-versioning'} === true) {
            return true;
        }
        return false;
    }
1132518
private void validateSchema(Schema schema) throws DDFException {
    Set<String> columnSet = new HashSet<String>();
    if(schema != null && schema.getColumns() != null) {
      for (Column column : schema.getColumns()) {
        if (columnSet.contains(column.getName())) {
          throw new DDFException(String.format("Duplicated column name %s", column.getName()));
        } else {
          columnSet.add(column.getName());
        }
      }
    }
  }
896406
private function getCreatedViews(Schema $oldSchema, Schema $newSchema)
    {
        $createdViews = array();

        foreach ($newSchema->getViews() as $view) {
            if (!$oldSchema->hasView($view->getName())
                || $this->compareViews($oldSchema->getView($view->getName()), $view)) {
                $createdViews[] = $view;
            }
        }

        return $createdViews;
    }
913163
void comparePrimitiveTypes(Schema oldSchema,
                                                    Schema newSchema,
                                                    List<Message> messages,
                                                    String name) {
        if(oldSchema == null || newSchema == null) {
            throw new IllegalArgumentException("Old schema must both be a primitive type. Name="
                                               + name + ". Type=" + oldSchema);
        }

        Schema.Type oldType = oldSchema.getType();
        Schema.Type newType = newSchema.getType();

        if(oldType != newType) {
            if(((oldType == Schema.Type.INT && (newType == Schema.Type.LONG
                                                || newType == Schema.Type.FLOAT || newType == Schema.Type.DOUBLE))
                || (oldType == Schema.Type.LONG && (newType == Schema.Type.FLOAT || newType == Schema.Type.DOUBLE)) || (oldType == Schema.Type.FLOAT && (newType == Schema.Type.DOUBLE)))) {
                messages.add(new Message(Level.INFO, "Type change from " + oldSchema.getType()
                                                     + " to " + newSchema.getType() + " for field "
                                                     + name));
            } else {
                messages.add(new Message(Level.ERROR, "Illegal type change from "
                                                      + oldSchema.getType() + " to "
                                                      + newSchema.getType() + " for field " + name));
            }
        }
    }
831217
@Override
    public Schema transform(Schema schema) {
        int nCols = schema.numColumns();
        List<ColumnMetaData> meta = schema.getColumnMetaData();
        List<ColumnMetaData> newMeta = new ArrayList<>(nCols);
        newMeta.addAll(meta);
        newMeta.add(new StringMetaData(outputColumnName));
        return schema.newSchema(newMeta);
    }