gora-elasticsearch/src/main/java/org/apache/gora/elasticsearch/mapping/ElasticsearchMappingBuilder.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.gora.elasticsearch.mapping;

import com.google.inject.ConfigurationException;
import org.apache.commons.io.IOUtils;
import org.apache.gora.elasticsearch.store.ElasticsearchStore;
import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.util.GoraException;
import org.jdom.Document;
import org.jdom.Element;
import org.jdom.JDOMException;
import org.jdom.input.SAXBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xml.sax.SAXException;

import javax.xml.XMLConstants;
import javax.xml.transform.Source;
import javax.xml.transform.stream.StreamSource;
import javax.xml.validation.Schema;
import javax.xml.validation.SchemaFactory;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Locale;

/**
 * Builder for Mapping definitions of Elasticsearch.
 */
public class ElasticsearchMappingBuilder<K, T extends PersistentBase> {

    private static final Logger LOG = LoggerFactory.getLogger(ElasticsearchMappingBuilder.class);

    /**
     * XSD validation file for the XML mapping.
     */
    private static final String XSD_MAPPING_FILE = "gora-elasticsearch.xsd";

    // Index description
    static final String ATT_NAME = "name";

    static final String ATT_TYPE = "type";

    // Class description
    static final String TAG_CLASS = "class";

    static final String ATT_KEYCLASS = "keyClass";

    static final String ATT_INDEX = "index";

    static final String TAG_FIELD = "field";

    static final String ATT_DOCFIELD = "docfield";

    static final String ATT_SCALINGFACTOR = "scalingFactor";

    /**
     * Mapping instance being built.
     */
    private ElasticsearchMapping elasticsearchMapping;

    private final ElasticsearchStore<K, T> dataStore;

    /**
     * Constructor for ElasticsearchMappingBuilder.
     *
     * @param store ElasticsearchStore instance
     */
    public ElasticsearchMappingBuilder(final ElasticsearchStore<K, T> store) {
        this.elasticsearchMapping = new ElasticsearchMapping();
        this.dataStore = store;
    }

    /**
     * Returns the Elasticsearch Mapping being built.
     *
     * @return Elasticsearch Mapping instance
     */
    public ElasticsearchMapping getElasticsearchMapping() {
        return elasticsearchMapping;
    }

    /**
     * Sets the Elasticsearch Mapping.
     *
     * @param elasticsearchMapping Elasticsearch Mapping instance
     */
    public void setElasticsearchMapping(ElasticsearchMapping elasticsearchMapping) {
        this.elasticsearchMapping = elasticsearchMapping;
    }

    /**
     * Reads Elasticsearch mappings from file.
     *
     * @param inputStream   Mapping input stream
     * @param xsdValidation Parameter for enabling XSD validation
     */
    public void readMappingFile(InputStream inputStream, boolean xsdValidation) {
        try {
            SAXBuilder saxBuilder = new SAXBuilder();
            if (inputStream == null) {
                LOG.error("The mapping input stream is null!");
                throw new GoraException("The mapping input stream is null!");
            }

            // Convert input stream to a string to use it a few times
            String mappingStream = IOUtils.toString(inputStream, Charset.defaultCharset());

            // XSD validation for XML file
            if (xsdValidation) {
                Source xmlSource = new StreamSource(IOUtils.toInputStream(mappingStream, Charset.defaultCharset()));
                Schema schema = SchemaFactory.newInstance(XMLConstants.W3C_XML_SCHEMA_NS_URI)
                        .newSchema(new StreamSource(getClass().getClassLoader().getResourceAsStream(XSD_MAPPING_FILE)));
                schema.newValidator().validate(xmlSource);
                LOG.info("Mapping file is valid.");
            }

            Document document = saxBuilder.build(IOUtils.toInputStream(mappingStream, Charset.defaultCharset()));
            if (document == null) {
                LOG.error("The mapping document is null!");
                throw new GoraException("The mapping document is null!");
            }

            Element root = document.getRootElement();
            // Extract class descriptions
            @SuppressWarnings("unchecked")
            List<Element> classElements = root.getChildren(TAG_CLASS);
            for (Element classElement : classElements) {
                final Class<T> persistentClass = dataStore.getPersistentClass();
                final Class<K> keyClass = dataStore.getKeyClass();
                if (haveKeyClass(keyClass, classElement)
                        && havePersistentClass(persistentClass, classElement)) {
                    loadPersistentClass(classElement, persistentClass);
                    break;
                }
            }
        } catch (IOException | JDOMException | ConfigurationException | SAXException ex) {
            throw new RuntimeException(ex);
        }
        LOG.info("Gora Elasticsearch mapping file was read successfully.");
    }

    private boolean haveKeyClass(final Class<K> keyClass,
                                 final Element classElement) {
        return classElement.getAttributeValue(ATT_KEYCLASS).equals(
                keyClass.getName());
    }

    private boolean havePersistentClass(final Class<T> persistentClass,
                                        final Element classElement) {
        return classElement.getAttributeValue(ATT_NAME).equals(
                persistentClass.getName());
    }

    /**
     * Handle the XML parsing of the class definition.
     *
     * @param classElement the XML node containing the class definition
     */
    protected void loadPersistentClass(Element classElement,
                                       Class<T> pPersistentClass) {

        String indexNameFromMapping = classElement.getAttributeValue(ATT_INDEX);
        String indexName = dataStore.getSchemaName(indexNameFromMapping, pPersistentClass);

        elasticsearchMapping.setIndexName(indexName);
        // docNameFromMapping could be null here
        if (!indexName.equals(indexNameFromMapping)) {
            ElasticsearchStore.LOG
                    .info("Keyclass and nameclass match, but mismatching index names. "
                            + "Mappingfile schema is '{}' vs actual schema '{}', assuming they are the same.",
                            indexNameFromMapping, indexName);
            if (indexNameFromMapping != null) {
                elasticsearchMapping.setIndexName(indexName);
            }
        }

        // Process fields declaration
        @SuppressWarnings("unchecked")
        List<Element> fields = classElement.getChildren(TAG_FIELD);
        for (Element fieldElement : fields) {
            String fieldTypeName = fieldElement.getAttributeValue(ATT_TYPE).toUpperCase(Locale.getDefault());
            Field.FieldType fieldType = new Field.FieldType(Field.DataType.valueOf(fieldTypeName));
            Field field;
            if (fieldType.getType() == Field.DataType.SCALED_FLOAT) {
                int scalingFactor = Integer.parseInt(fieldElement.getAttributeValue(ATT_SCALINGFACTOR));
                field = new Field(fieldElement.getAttributeValue(ATT_DOCFIELD), new Field.FieldType(scalingFactor));
            } else {
                field = new Field(fieldElement.getAttributeValue(ATT_DOCFIELD), fieldType);
            }
            elasticsearchMapping.addField(fieldElement.getAttributeValue(ATT_NAME), field);
        }
    }
}