KunderaExport.java

public class KunderaExport {

  public static void main(String... args) {
		generateDdl();
	}

	private void generateDdl() throws Exception {
		getLog().info("Processing persistence unit: " + persistenceUnitName);

		// instantiate app metadata
		ApplicationMetadata appMetadata = KunderaMetadata.INSTANCE
		        .getApplicationMetadata();

		// read persistence-units from persistence.xml
		List<PersistenceUnitMetadata> metadatas = new ArrayList<PersistenceUnitMetadata>();
		Enumeration<URL> xmls = Thread.currentThread().getContextClassLoader()
		        .getResources("META-INF/persistence.xml");
		while (xmls.hasMoreElements())
			metadatas = PersistenceXMLLoader.findPersistenceUnits(xmls
			        .nextElement());
		getLog().debug("Total PUs: " + metadatas.size());

		// pu to schema metadata map
		puToSchemaMetadata = appMetadata.getSchemaMetadata()
		        .getPuToSchemaMetadata();
		// add persistence-units metadata to app metadata
		for (PersistenceUnitMetadata pum : metadatas) {
			String persistenceUnit = pum.getPersistenceUnitName();
			getLog().info("Processing persistence-unit: " + persistenceUnit);

			String keyspaceNameProperty = pum.getProperties().getProperty(
			        "kundera.keyspace");
			keyspaceName = keyspaceNameProperty == null ? keyspaceName
			        : keyspaceNameProperty;

			// initialize app metadata
			Map<String, PersistenceUnitMetadata> metadata = new HashMap<String, PersistenceUnitMetadata>();
			metadata.put(persistenceUnit, pum);
			appMetadata.addPersistenceUnitMetadata(metadata);

			// populate persistence-unit entity metadata
			TableProcessor processor = new TableProcessor(pum.getProperties());
			// iterate all classes defined for this persistence-unit
			getLog().debug("Processing " + pum.getClasses().size() + " classes");
			for (String className : pum.getManagedClassNames()) {
				getLog().debug("Processing class: " + className);
				Class clazz = pum.getClassLoader().loadClass(className);

				// process class
				EntityMetadata em = new EntityMetadata(clazz);
				processor.process(clazz, em);
				em.setPersistenceUnit(persistenceUnit);
				MetamodelImpl metaModel = new MetamodelImpl();
				metaModel.addEntityMetadata(clazz, em);

				// add class metamodel to app metadata
				appMetadata.getMetamodelMap().put(persistenceUnit, metaModel);
				metaModel.assignManagedTypes(appMetadata.getMetaModelBuilder(
				        persistenceUnit).getManagedTypes());
				metaModel.assignEmbeddables(appMetadata.getMetaModelBuilder(
				        persistenceUnit).getEmbeddables());
				metaModel.assignMappedSuperClass(appMetadata
				        .getMetaModelBuilder(persistenceUnit)
				        .getMappedSuperClassTypes());
				appMetadata.getMetamodelMap().put(persistenceUnit, metaModel);
			}

			// retrieve processed classes metadata
			Map<Class<?>, EntityMetadata> entityMetadataMap = getEntityMetadataCol(
			        appMetadata, persistenceUnit);

			// retrieve configured tables and columns
			List<TableInfo> tableInfos = getSchemaInfo(persistenceUnit);

			// Iterate each entity metadata.
			for (EntityMetadata entityMetadata : entityMetadataMap.values()) {
				// get entity metadata(table info as well as columns)
				// if table info exists, get it from map.
				boolean found = false;
				Type type = entityMetadata.getType();
				Class idClassName = entityMetadata.getIdAttribute() != null ? entityMetadata
				        .getIdAttribute().getJavaType() : null;
				TableInfo tableInfo = new TableInfo(
				        entityMetadata.getTableName(),
				        entityMetadata.isIndexable(), type.name(), idClassName);

				// check for tableInfos not empty and contains the present
				// tableInfo.
				if (!tableInfos.isEmpty() && tableInfos.contains(tableInfo)) {
					found = true;
					int idx = tableInfos.indexOf(tableInfo);
					tableInfo = tableInfos.get(idx);
					addColumnToTableInfo(entityMetadata, type, tableInfo);
				} else
					addColumnToTableInfo(entityMetadata, type, tableInfo);

				List<Relation> relations = entityMetadata.getRelations();
				parseRelations(persistenceUnit, tableInfos, entityMetadata,
				        tableInfo, relations);

				if (!found)
					tableInfos.add(tableInfo);

				// Add table for GeneratedValue if opted TableStrategy
				addTableGenerator(appMetadata, persistenceUnit, tableInfos,
				        entityMetadata, idClassName);

			}
			puToSchemaMetadata.put(persistenceUnit, tableInfos);
		}

		// get persistence-unit metadata
		Map<String, List<TableInfo>> pus = appMetadata.getSchemaMetadata()
		        .getPuToSchemaMetadata(); // puSchema
		PersistenceUnitMetadata puMetadata = KunderaMetadata.INSTANCE
		        .getApplicationMetadata().getPersistenceUnitMetadata(
		                persistenceUnitName);

		// traverse tables and process CfDef
		List<CfDef> cfDefs = new ArrayList<CfDef>();
		for (TableInfo tableInfo : pus.get(persistenceUnitName)) {
			getLog().info("TableInfo: " + tableInfo.toString());
			if (tableInfo.getTableIdType() != null
			        && tableInfo.getTableIdType().isAnnotationPresent(
			                Embeddable.class)) {
				if (tableInfo.getType() != null
				        && tableInfo.getType().equals(
				                Type.SUPER_COLUMN_FAMILY.name()))
					throw new SchemaGenerationException(
					        "Composite/Compound columns are yet supported over Super column family by Cassandra",
					        "cassandra", puMetadata
					                .getProperty("kundera.keyspace"));
				else
					onCompoundKey(tableInfo);
			} else {
				cfDefs.add(getTableMetadata(tableInfo));
				// Create Index Table if required
				CfDef possibleInvertedIndexTable = getInvertedIndexCF(tableInfo);
				if (possibleInvertedIndexTable != null)
					cfDefs.add(possibleInvertedIndexTable);
			}
		}

		// configure keyspace
		// TODO placementStrategy should be read from somewhere (plug-in conf,
		// properties file, ..)
		String placementStrategy = SimpleStrategy.class.getName();
		KsDef ksDef = new KsDef(keyspaceName, placementStrategy, cfDefs);

		// options should be read as well (right now, I have this on YAML file)
		Map<String, String> ksOptions = new HashMap<String, String>();

		// replication factor only settable if using SimpleStrategy/none
		// defined. see CassandraSchemaManager:957/977
		ksDef.setReplication_factor(1);
		ksOptions.put(CassandraConstants.REPLICATION_FACTOR, "1");

		// TODO define datacenters if using NetworkTopologyStrategy. see
		// CassandraSchemaManager:966

		// TODO define durable_writes
		ksOptions.put(CassandraConstants.DURABLE_WRITES, "false");

		// set options in keyspace
		ksDef.setStrategy_options(ksOptions);

		// create keyspace and add cfdefs
		export(createScriptFileName, ksDef.toString());
	}

	/**
	 * Returns list of configured table/column families.
	 * 
	 * @param persistenceUnit
	 *            persistence unit, for which schema needs to be fetched.
	 * 
	 * @return list of {@link TableInfo}
	 */
	private List<TableInfo> getSchemaInfo(String persistenceUnit) {
		List<TableInfo> tableInfos = puToSchemaMetadata.get(persistenceUnit);
		// if no TableInfos for given persistence unit.
		if (tableInfos == null) {
			tableInfos = new ArrayList<TableInfo>();
		}
		return tableInfos;
	}

	/**
	 * Returns map of entity metdata {@link EntityMetadata}.
	 * 
	 * @param appMetadata
	 *            application metadata
	 * @param persistenceUnit
	 *            persistence unit
	 * @return map of entity metadata.
	 */
	private Map<Class<?>, EntityMetadata> getEntityMetadataCol(
	        ApplicationMetadata appMetadata, String persistenceUnit) {
		Metamodel metaModel = appMetadata.getMetamodel(persistenceUnit);
		Map<Class<?>, EntityMetadata> entityMetadataMap = ((MetamodelImpl) metaModel)
		        .getEntityMetadataMap();
		return entityMetadataMap;
	}

	private EmbeddedColumnInfo getEmbeddedColumn(EmbeddableType embeddableType,
	        String embeddableColName, Class embeddedEntityClass) {
		EmbeddedColumnInfo embeddedColumnInfo = new EmbeddedColumnInfo(
		        embeddableType);
		embeddedColumnInfo.setEmbeddedColumnName(embeddableColName);
		Map<String, PropertyIndex> indexedColumns = IndexProcessor
		        .getIndexesOnEmbeddable(embeddedEntityClass);
		List<ColumnInfo> columns = new ArrayList<ColumnInfo>();

		Set attributes = embeddableType.getAttributes();
		Iterator<Attribute> iter = attributes.iterator();

		while (iter.hasNext()) {
			Attribute attr = iter.next();
			columns.add(getColumn(attr, indexedColumns.get(attr.getName())));
		}
		embeddedColumnInfo.setColumns(columns);
		return embeddedColumnInfo;
	}

	/**
	 * getColumn method return ColumnInfo for the given column
	 * 
	 * @param Object
	 *            of Column.
	 * @return Object of ColumnInfo.
	 */
	private ColumnInfo getColumn(Attribute column, PropertyIndex indexedColumn) {
		ColumnInfo columnInfo = new ColumnInfo();
		columnInfo.setColumnName(((AbstractAttribute) column)
		        .getJPAColumnName());
		if (indexedColumn != null && indexedColumn.getName() != null) {
			columnInfo.setIndexable(true);
			columnInfo.setIndexType(indexedColumn.getIndexType());
			columnInfo.setMaxValue(indexedColumn.getMax());
			columnInfo.setMinValue(indexedColumn.getMin());

			// Add more if required
		}
		columnInfo.setType(column.getJavaType());
		return columnInfo;
	}

	/**
	 * getJoinColumn method return ColumnInfo for the join column
	 * 
	 * @param String
	 *            joinColumnName.
	 * @return ColumnInfo object columnInfo.
	 */
	private ColumnInfo getJoinColumn(String joinColumnName) {
		ColumnInfo columnInfo = new ColumnInfo();
		columnInfo.setColumnName(joinColumnName);
		columnInfo.setIndexable(true);
		return columnInfo;
	}

	/**
	 * parse the relations of entites .
	 * 
	 * @param persistenceUnit
	 * @param tableInfos
	 * @param entityMetadata
	 * @param tableInfo
	 * @param relations
	 */
	private void parseRelations(String persistenceUnit,
	        List<TableInfo> tableInfos, EntityMetadata entityMetadata,
	        TableInfo tableInfo, List<Relation> relations) {
		for (Relation relation : relations) {
			Class entityClass = relation.getTargetEntity();
			EntityMetadata targetEntityMetadata = KunderaMetadataManager
			        .getEntityMetadata(entityClass);
			ForeignKey relationType = relation.getType();

			// if relation type is one to many or join by primary key
			if (targetEntityMetadata != null
			        && relationType.equals(ForeignKey.ONE_TO_MANY)
			        && relation.getJoinColumnName() != null) {
				// if self association
				if (targetEntityMetadata.equals(entityMetadata)) {
					tableInfo.addColumnInfo(getJoinColumn(relation
					        .getJoinColumnName()));
				} else {
					String pu = targetEntityMetadata.getPersistenceUnit();
					Type targetEntityType = targetEntityMetadata.getType();
					// Class idClass =
					// targetEntityMetadata.getIdColumn().getField().getType();
					Class idClass = targetEntityMetadata.getIdAttribute()
					        .getJavaType();
					TableInfo targetTableInfo = new TableInfo(
					        targetEntityMetadata.getTableName(),
					        targetEntityMetadata.isIndexable(),
					        targetEntityType.name(), idClass);

					// In case of different persistence unit. case for poly glot
					// persistence.
					if (!pu.equals(persistenceUnit)) {
						List<TableInfo> targetTableInfos = getSchemaInfo(pu);

						addJoinColumnToInfo(relation.getJoinColumnName(),
						        targetTableInfo, targetTableInfos);

						// add for newly discovered persistence unit.
						puToSchemaMetadata.put(pu, targetTableInfos);
					} else {
						addJoinColumnToInfo(relation.getJoinColumnName(),
						        targetTableInfo, tableInfos);
						// tableInfos.add(targetTableInfo);
					}
				}
			}
			// if relation type is one to one or many to one.
			else if (relation.isUnary() && relation.getJoinColumnName() != null) {
				tableInfo.addColumnInfo(getJoinColumn(relation
				        .getJoinColumnName()));
			}
			// if relation type is many to many and relation via join table.
			else if ((relationType.equals(ForeignKey.MANY_TO_MANY))
			        && (entityMetadata.isRelationViaJoinTable())) {
				JoinTableMetadata joinTableMetadata = relation
				        .getJoinTableMetadata();
				String joinTableName = joinTableMetadata != null ? joinTableMetadata
				        .getJoinTableName() : null;
				if (joinTableName != null) {
					TableInfo joinTableInfo = new TableInfo(joinTableName,
					        false, Type.COLUMN_FAMILY.name(), null);
					if (!tableInfos.isEmpty()
					        && !tableInfos.contains(joinTableInfo)
					        || tableInfos.isEmpty()) {
						tableInfos.add(joinTableInfo);
					}
				}
			}
		}
	}

	/**
	 * adds join column name to the table Info of entity.
	 * 
	 * @param joinColumn
	 * @param targetTableInfo
	 * @param targetTableInfos
	 */
	private void addJoinColumnToInfo(String joinColumn,
	        TableInfo targetTableInfo, List<TableInfo> targetTableInfos) {
		if (!targetTableInfos.isEmpty()
		        && targetTableInfos.contains(targetTableInfo)) {
			int idx = targetTableInfos.indexOf(targetTableInfo);
			targetTableInfo = targetTableInfos.get(idx);
			if (!targetTableInfo.getColumnMetadatas().contains(
			        getJoinColumn(joinColumn))) {
				targetTableInfo.addColumnInfo(getJoinColumn(joinColumn));
			}
		} else {
			if (!targetTableInfo.getColumnMetadatas().contains(
			        getJoinColumn(joinColumn))) {
				targetTableInfo.addColumnInfo(getJoinColumn(joinColumn));
			}
			targetTableInfos.add(targetTableInfo);
		}

		// targetTableInfo.addColumnInfo(getColumn(entityMetadata.getIdColumn()));
		// targetTableInfos.add(targetTableInfo);
	}

	/**
	 * Adds column to table info of entity.
	 * 
	 * @param entityMetadata
	 * @param type
	 * @param tableInfo
	 */
	private void addColumnToTableInfo(EntityMetadata entityMetadata, Type type,
	        TableInfo tableInfo) {
		// Add columns to table info.
		Metamodel metaModel = KunderaMetadata.INSTANCE.getApplicationMetadata()
		        .getMetamodel(entityMetadata.getPersistenceUnit());
		EntityType entityType = metaModel.entity(entityMetadata
		        .getEntityClazz());
		// List<IndexedColumn> columns = getIndexDefs(entityMetadata);
		// Map<String, com.impetus.kundera.newannotations.Index> columns =
		// entityMetadata.getColToBeIndexed();
		Map<String, PropertyIndex> columns = entityMetadata
		        .getIndexProperties();

		Set attributes = entityType.getAttributes();

		Iterator<Attribute> iter = attributes.iterator();
		while (iter.hasNext()) {
			Attribute attr = iter.next();
			if (!attr.isAssociation()) {
				if (((MetamodelImpl) metaModel)
				        .isEmbeddable(attr.getJavaType())) {
					EmbeddableType embeddable = metaModel.embeddable(attr
					        .getJavaType());

					EmbeddedColumnInfo embeddedColumnInfo = getEmbeddedColumn(
					        embeddable, attr.getName(), attr.getJavaType());

					if (!tableInfo.getEmbeddedColumnMetadatas().contains(
					        embeddedColumnInfo)) {
						tableInfo.addEmbeddedColumnInfo(embeddedColumnInfo);
					}
				} else if (!attr.isCollection()
				        && !((SingularAttribute) attr).isId()) {
					ColumnInfo columnInfo = getColumn(
					        attr,
					        columns != null ? columns
					                .get(((AbstractAttribute) attr)
					                        .getJPAColumnName()) : null);
					if (!tableInfo.getColumnMetadatas().contains(columnInfo)) {
						tableInfo.addColumnInfo(columnInfo);
					}
				}
			}
		}
	}

	private void addTableGenerator(ApplicationMetadata appMetadata,
	        String persistenceUnit, List<TableInfo> tableInfos,
	        EntityMetadata entityMetadata, Class idClassName) {
		Metamodel metamodel = appMetadata.getMetamodel(persistenceUnit);
		IdDiscriptor keyValue = ((MetamodelImpl) metamodel)
		        .getKeyValue(entityMetadata.getEntityClazz().getName());
		if (keyValue != null && keyValue.getTableDiscriptor() != null) {
			TableInfo tableGeneratorDiscriptor = new TableInfo(keyValue
			        .getTableDiscriptor().getTable(), false,
			        "CounterColumnType", idClassName);
			if (!tableInfos.contains(tableGeneratorDiscriptor)) {
				tableInfos.add(tableGeneratorDiscriptor);
			}
		}
	}

	private void export(String outFile, String data) {
		outputDirectory.mkdirs();
		File file = new File(outputDirectory, outFile);
		try {
			PrintWriter writer = new PrintWriter(file, Charset
			        .forName(encoding).name());
			try {
				writer.print(data);
				getLog().info("wrote: " + file);
			} finally {
				writer.close();
			}
		} catch (Exception e) {
			getLog().error(e.toString(), e);
			throw new RuntimeException(e);
		}
	}

	/**
	 * On compound key.
	 * 
	 * @param tableInfo
	 *            the table info
	 * @throws InvalidRequestException
	 *             the invalid request exception
	 * @throws TException
	 *             the t exception
	 * @throws SchemaDisagreementException
	 *             the schema disagreement exception
	 */
	private void onCompoundKey(TableInfo tableInfo)
	        throws InvalidRequestException, TException,
	        SchemaDisagreementException {
		CQLTranslator translator = new CQLTranslator();
		String columnFamilyQuery = CQLTranslator.CREATE_COLUMNFAMILY_QUERY;
		columnFamilyQuery = StringUtils.replace(
		        columnFamilyQuery,
		        CQLTranslator.COLUMN_FAMILY,
		        translator.ensureCase(new StringBuilder(),
		                tableInfo.getTableName()).toString());

		List<ColumnInfo> columns = tableInfo.getColumnMetadatas();

		StringBuilder queryBuilder = new StringBuilder();

		// for normal columns
		onCompositeColumns(translator, columns, queryBuilder);

		// ideally it will always be one as more super column families
		// are not allowed with compound/composite key.

		List<EmbeddedColumnInfo> compositeColumns = tableInfo
		        .getEmbeddedColumnMetadatas();
		EmbeddableType compoEmbeddableType = compositeColumns.get(0)
		        .getEmbeddable();

		// for composite columns
		onCompositeColumns(translator, compositeColumns.get(0).getColumns(),
		        queryBuilder);

		// strip last ",".
		if (queryBuilder.length() > 0) {
			queryBuilder.deleteCharAt(queryBuilder.length() - 1);

			columnFamilyQuery = StringUtils.replace(columnFamilyQuery,
			        CQLTranslator.COLUMNS, queryBuilder.toString());
			queryBuilder = new StringBuilder(columnFamilyQuery);
		}

		// append primary key clause

		queryBuilder.append(CQLTranslator.ADD_PRIMARYKEY_CLAUSE);

		Field[] fields = tableInfo.getTableIdType().getDeclaredFields();

		StringBuilder primaryKeyBuilder = new StringBuilder();

		for (Field f : fields) {
			if (!ReflectUtils.isTransientOrStatic(f)) {
				Attribute attribute = compoEmbeddableType.getAttribute(f
				        .getName());
				translator.appendColumnName(primaryKeyBuilder,
				        ((AbstractAttribute) attribute).getJPAColumnName());
				primaryKeyBuilder.append(" ,");
			}
		}

		// should not be null.
		primaryKeyBuilder.deleteCharAt(primaryKeyBuilder.length() - 1);

		queryBuilder = new StringBuilder(StringUtils.replace(
		        queryBuilder.toString(), CQLTranslator.COLUMNS,
		        primaryKeyBuilder.toString()));

		// set column family properties defined in configuration property/xml
		// files.
		setColumnFamilyProperties(null, getColumnFamilyProperties(tableInfo),
		        queryBuilder);
	}

	/**
	 * On composite columns.
	 * 
	 * @param translator
	 *            the translator
	 * @param columns
	 *            the columns
	 * @param queryBuilder
	 *            the query builder
	 */
	private void onCompositeColumns(CQLTranslator translator,
	        List<ColumnInfo> columns, StringBuilder queryBuilder) {
		for (ColumnInfo colInfo : columns) {
			String dataType = CassandraValidationClassMapper
			        .getValidationClass(colInfo.getType());
			String cqlType = CQLTranslator.getCQLType(dataType);
			translator.appendColumnName(queryBuilder, colInfo.getColumnName(),
			        cqlType);
			queryBuilder.append(" ,");
		}
	}

	/**
	 * @param tableInfo
	 * @throws InvalidRequestException
	 * @throws SchemaDisagreementException
	 * @throws TException
	 */
	private CfDef getInvertedIndexCF(TableInfo tableInfo)
	        throws InvalidRequestException, SchemaDisagreementException,
	        TException {
		boolean indexTableRequired = tableInfo.getEmbeddedColumnMetadatas()
		        .isEmpty();
		if (indexTableRequired) {
			CfDef cfDef = new CfDef();
			cfDef.setKeyspace(keyspaceName);
			cfDef.setColumn_type("Super");
			cfDef.setName(tableInfo.getTableName()
			        + Constants.INDEX_TABLE_SUFFIX);
			cfDef.setKey_validation_class(UTF8Type.class.getSimpleName());
			return cfDef;
		}
		return null;
	}

	/**
	 * get Table metadata method returns the metadata of table for given
	 * tableInfo.
	 * 
	 * @param tableInfo
	 *            the table info
	 * @return the table metadata
	 */
	/**
	 * @param tableInfo
	 * @return CfDef object
	 */
	private CfDef getTableMetadata(TableInfo tableInfo) {
		CfDef cfDef = new CfDef();
		cfDef.setKeyspace(keyspaceName);
		cfDef.setName(tableInfo.getTableName());
		cfDef.setKey_validation_class(CassandraValidationClassMapper
		        .getValidationClass(tableInfo.getTableIdType()));

		Properties cFProperties = getColumnFamilyProperties(tableInfo);
		String defaultValidationClass = null;
		if (tableInfo.getType() != null
		        && tableInfo.getType().equals(Type.SUPER_COLUMN_FAMILY.name())) {
			if (isCounterColumnType(tableInfo, defaultValidationClass)) {
				cfDef.setDefault_validation_class(CounterColumnType.class
				        .getSimpleName());
			}
			cfDef.setColumn_type("Super");
			cfDef.setComparator_type(UTF8Type.class.getSimpleName());
			cfDef.setSubcomparator_type(UTF8Type.class.getSimpleName());
		} else if (tableInfo.getType() != null) {
			defaultValidationClass = cFProperties != null ? cFProperties
			        .getProperty(CassandraConstants.DEFAULT_VALIDATION_CLASS)
			        : null;
			cfDef.setColumn_type("Standard");
			cfDef.setComparator_type(UTF8Type.class.getSimpleName());
			if (isCounterColumnType(tableInfo, defaultValidationClass)) {
				cfDef.setDefault_validation_class(CounterColumnType.class
				        .getSimpleName());
			} else {
				List<ColumnDef> columnDefs = new ArrayList<ColumnDef>();
				List<ColumnInfo> columnInfos = tableInfo.getColumnMetadatas();
				if (columnInfos != null) {
					for (ColumnInfo columnInfo : columnInfos) {
						ColumnDef columnDef = new ColumnDef();
						if (columnInfo.isIndexable()) {
							columnDef.setIndex_type(CassandraIndexHelper
							        .getIndexType(columnInfo.getIndexType()));
						}
						columnDef
						        .setName(columnInfo.getColumnName().getBytes());
						columnDef
						        .setValidation_class(CassandraValidationClassMapper
						                .getValidationClass(columnInfo
						                        .getType()));
						columnDefs.add(columnDef);
					}
				}
				cfDef.setColumn_metadata(columnDefs);
			}
		}
		setColumnFamilyProperties(cfDef, cFProperties, null);
		return cfDef;
	}

	/**
	 * Checks if is counter column type.
	 * 
	 * @param tableInfo
	 *            the table info
	 * @param defaultValidationClass
	 *            the default validation class
	 * @return true, if is counter column type
	 */
	private boolean isCounterColumnType(TableInfo tableInfo,
	        String defaultValidationClass) {
		return (defaultValidationClass != null
		        && (defaultValidationClass
		                .equalsIgnoreCase(CounterColumnType.class
		                        .getSimpleName()) || defaultValidationClass
		                .equalsIgnoreCase(CounterColumnType.class.getName())) || (tableInfo
		            .getType().equals(CounterColumnType.class.getSimpleName())));
	}

	/**
	 * Gets the column family properties.
	 * 
	 * @param tableInfo
	 *            the table info
	 * @return the column family properties
	 */
	private Properties getColumnFamilyProperties(TableInfo tableInfo) {
		if (tables != null) {
			for (Table table : tables) {
				if (table != null
				        && table.getName() != null
				        && table.getName().equalsIgnoreCase(
				                tableInfo.getTableName())) {
					return table.getProperties();
				}
			}
		}
		return null;
	}

	/**
	 * Enum ColumnFamilyType for type of column family in cassandra ie Super or
	 * Standard.
	 * 
	 */
	private enum ColumnFamilyType {
		/** The Standard. */
		Standard,
		/** The Super. */
		Super;
	}

	/**
	 * Sets the column family properties.
	 * 
	 * @param cfDef
	 *            the cf def
	 * @param cFProperties
	 *            the c f properties
	 */
	private void setColumnFamilyProperties(CfDef cfDef,
	        Properties cFProperties, StringBuilder builder) {
		if ((cfDef != null && cFProperties != null)
		        || (builder != null && cFProperties != null)) {
			if (builder != null) {

				builder.append(CQLTranslator.WITH_CLAUSE);
			}
			String keyValidationClass = cFProperties
			        .getProperty(CassandraConstants.KEY_VALIDATION_CLASS);
			if (keyValidationClass != null) {
				if (builder != null) {
					// nothing available.
				} else {
					cfDef.setKey_validation_class(keyValidationClass);
				}
			}

			String compactionStrategy = cFProperties
			        .getProperty(CassandraConstants.COMPACTION_STRATEGY);
			if (compactionStrategy != null) {
				if (builder != null) {
					String strategy_class = CQLTranslator
					        .getKeyword(CassandraConstants.COMPACTION_STRATEGY);
					builder.append(strategy_class);
					builder.append(CQLTranslator.EQ_CLAUSE);
					builder.append(CQLTranslator.QUOTE_STR);
					builder.append(compactionStrategy);
					builder.append(CQLTranslator.QUOTE_STR);
					builder.append(CQLTranslator.AND_CLAUSE);
				} else {
					cfDef.setCompaction_strategy(compactionStrategy);

				}
			}

			String comparatorType = cFProperties
			        .getProperty(CassandraConstants.COMPARATOR_TYPE);
			if (comparatorType != null) {
				if (builder != null) {
					// TODO:::nothing available.
				} else {
					cfDef.setComparator_type(comparatorType);

				}
			}

			String subComparatorType = cFProperties
			        .getProperty(CassandraConstants.SUBCOMPARATOR_TYPE);
			if (subComparatorType != null
			        && ColumnFamilyType.valueOf(cfDef.getColumn_type()) == ColumnFamilyType.Super) {
				if (builder != null) {
					// super column are not supported for composite key as of
					// now, leaving blank place holder..
				} else {
					cfDef.setSubcomparator_type(subComparatorType);
				}
			}
			String replicateOnWrite = cFProperties
			        .getProperty(CassandraConstants.REPLICATE_ON_WRITE);
			if (builder != null) {
				appendPropertyToBuilder(builder, replicateOnWrite,
				        CassandraConstants.REPLICATE_ON_WRITE);
			} else {
				cfDef.setReplicate_on_write(Boolean
				        .parseBoolean(replicateOnWrite));
			}
			String maxCompactionThreshold = cFProperties
			        .getProperty(CassandraConstants.MAX_COMPACTION_THRESHOLD);
			if (maxCompactionThreshold != null) {
				try {
					if (builder != null) {
						// Somehow these are not working for cassandra 1.1
						// though they claim it should work.
						// appendPropertyToBuilder(builder,
						// maxCompactionThreshold,
						// CassandraConstants.MAX_COMPACTION_THRESHOLD);
					} else {
						cfDef.setMax_compaction_threshold(Integer
						        .parseInt(maxCompactionThreshold));
					}
				} catch (NumberFormatException nfe) {
					getLog().error(
					        "Max_Compaction_Threshold should be numeric type");
					throw new SchemaGenerationException(nfe);
				}
			}
			String minCompactionThreshold = cFProperties
			        .getProperty(CassandraConstants.MIN_COMPACTION_THRESHOLD);
			if (minCompactionThreshold != null) {
				try {
					if (builder != null) {
						// Somehow these are not working for cassandra 1.1
						// though they claim it should work.
						// appendPropertyToBuilder(builder,
						// minCompactionThreshold,
						// CassandraConstants.MIN_COMPACTION_THRESHOLD);
					} else {
						cfDef.setMin_compaction_threshold(Integer
						        .parseInt(minCompactionThreshold));
					}
				} catch (NumberFormatException nfe) {
					getLog().error(
					        "Min_Compaction_Threshold should be numeric type");
					throw new SchemaGenerationException(nfe);
				}
			}

			String comment = cFProperties
			        .getProperty(CassandraConstants.COMMENT);
			if (comment != null) {
				if (builder != null) {
					String comment_Str = CQLTranslator
					        .getKeyword(CassandraConstants.COMMENT);
					builder.append(comment_Str);
					builder.append(CQLTranslator.EQ_CLAUSE);
					builder.append(CQLTranslator.QUOTE_STR);
					builder.append(comment);
					builder.append(CQLTranslator.QUOTE_STR);
					builder.append(CQLTranslator.AND_CLAUSE);

				} else {
					cfDef.setComment(comment);
				}
			}

			String id = cFProperties.getProperty(CassandraConstants.ID);
			if (id != null) {
				try {
					if (builder != null) {
						// TODO::::not available with composite key?
					} else {
						cfDef.setId(Integer.parseInt(id));
					}
				} catch (NumberFormatException nfe) {
					getLog().error("Id should be numeric type");
					throw new SchemaGenerationException(nfe);
				}
			}

			String gcGraceSeconds = cFProperties
			        .getProperty(CassandraConstants.GC_GRACE_SECONDS);
			if (gcGraceSeconds != null) {
				try {
					if (builder != null) {
						appendPropertyToBuilder(builder, gcGraceSeconds,
						        CassandraConstants.GC_GRACE_SECONDS);
					} else {
						cfDef.setGc_grace_seconds(Integer
						        .parseInt(gcGraceSeconds));
					}
				} catch (NumberFormatException nfe) {
					getLog().error("GC_GRACE_SECONDS should be numeric type");
					throw new SchemaGenerationException(nfe);
				}
			}

			String caching = cFProperties
			        .getProperty(CassandraConstants.CACHING);
			if (caching != null) {
				if (builder != null) {
					appendPropertyToBuilder(builder, caching,
					        CassandraConstants.CACHING);
				} else {
					cfDef.setCaching(caching);
				}
			}

			String bloomFilterFpChance = cFProperties
			        .getProperty(CassandraConstants.BLOOM_FILTER_FP_CHANCE);
			if (bloomFilterFpChance != null) {
				try {
					if (builder != null) {
						appendPropertyToBuilder(builder, bloomFilterFpChance,
						        CassandraConstants.BLOOM_FILTER_FP_CHANCE);
					} else {
						cfDef.setBloom_filter_fp_chance(Double
						        .parseDouble(bloomFilterFpChance));
					}
				} catch (NumberFormatException nfe) {
					getLog().error(
					        "BLOOM_FILTER_FP_CHANCE should be double type");
					throw new SchemaGenerationException(nfe);
				}
			}
			String readRepairChance = cFProperties
			        .getProperty(CassandraConstants.READ_REPAIR_CHANCE);
			if (readRepairChance != null) {
				try {
					if (builder != null) {
						appendPropertyToBuilder(builder, readRepairChance,
						        CassandraConstants.READ_REPAIR_CHANCE);
					} else {
						cfDef.setRead_repair_chance(Double
						        .parseDouble(readRepairChance));
					}
				} catch (NumberFormatException nfe) {
					getLog().error("READ_REPAIR_CHANCE should be double type");
					throw new SchemaGenerationException(nfe);
				}
			}
			String dclocalReadRepairChance = cFProperties
			        .getProperty(CassandraConstants.DCLOCAL_READ_REPAIR_CHANCE);
			if (dclocalReadRepairChance != null) {
				try {
					if (builder != null) {
						appendPropertyToBuilder(builder,
						        dclocalReadRepairChance,
						        CassandraConstants.DCLOCAL_READ_REPAIR_CHANCE);

					} else {
						cfDef.setDclocal_read_repair_chance(Double
						        .parseDouble(dclocalReadRepairChance));
					}
				} catch (NumberFormatException nfe) {
					getLog().error("READ_REPAIR_CHANCE should be double type");
					throw new SchemaGenerationException(nfe);
				}
			}

			// Strip last AND clause.
			if (builder != null
			        && StringUtils.contains(builder.toString(),
			                CQLTranslator.AND_CLAUSE)) {
				builder.delete(builder.lastIndexOf(CQLTranslator.AND_CLAUSE),
				        builder.length());
				// builder.deleteCharAt(builder.length() - 2);
			}
		}
	}

	/**
	 * @param builder
	 * @param replicateOnWrite
	 * @param keyword
	 */
	private void appendPropertyToBuilder(StringBuilder builder,
	        String replicateOnWrite, String keyword) {
		String replicateOn_Write = CQLTranslator.getKeyword(keyword);
		builder.append(replicateOn_Write);
		builder.append(CQLTranslator.EQ_CLAUSE);
		builder.append(replicateOnWrite);
		builder.append(CQLTranslator.AND_CLAUSE);
	}

	private ClassLoader getClassloader(ClassLoader old) {
		return new URLClassLoader(getCompileClassPath().toArray(new URL[0]),
		        old);
	}

	private List<URL> getCompileClassPath() {
		try {
			return Lists.transform(project.getCompileClasspathElements(),
			        new Function<String, URL>() {
				        @Override
				        public URL apply(String input) {
					        try {
						        return new File(input).toURI().toURL();
					        } catch (MalformedURLException e) {
						        throw new AssertionError(e);
					        }
				        }
			        });
		} catch (DependencyResolutionRequiredException e) {
			throw new AssertionError(e);
		}
	}

}