Skip to content

Commit

Permalink
Add 'version_type' support (implied when using version mapping)
Browse files Browse the repository at this point in the history 
relates to #343
  • Loading branch information
@costin
costin committed Jan 12, 2015
1 parent da911e1 commit a5148b8
Show file tree
Hide file tree
Showing 4 changed files with 52 additions and 17 deletions.
15 changes: 12 additions & 3 deletions mr/src/itest/java/org/elasticsearch/hadoop/integration/mr/AbstractMROldApiSaveTest.java
View file
Expand Up @@ -411,6 +411,15 @@ public void testIndexAutoCreateDisabled() throws Exception {
runJob(conf);
}

@Test
public void testIndexWithVersionMappingImpliesVersionTypeExternal() throws Exception {
JobConf conf = createJobConf();
conf.set(ConfigurationOptions.ES_RESOURCE, "mroldapi/external-version-implied");
conf.set(ConfigurationOptions.ES_MAPPING_VERSION, "number");

runJob(conf);
}

@Test
public void testParentChild() throws Exception {
JobConf conf = createJobConf();
Expand All @@ -422,7 +431,7 @@ public void testParentChild() throws Exception {
runJob(conf);
}

@Test
@Test
public void testIndexPattern() throws Exception {
JobConf conf = createJobConf();
conf.set(ConfigurationOptions.ES_RESOURCE, "/mroldapi/pattern-{number}");
Expand All @@ -431,7 +440,7 @@ public void testIndexPattern() throws Exception {
runJob(conf);
}

@Test
@Test
public void testIndexPatternWithFormatting() throws Exception {
JobConf conf = createJobConf();
conf.set(ConfigurationOptions.ES_RESOURCE, "mroldapi/pattern-format-{@timestamp:YYYY-MM-dd}");
Expand All @@ -440,7 +449,7 @@ public void testIndexPatternWithFormatting() throws Exception {
runJob(conf);
}

@Test
@Test
public void testIndexPatternWithFormattingAndId() throws Exception {
JobConf conf = createJobConf();
conf.set(ConfigurationOptions.ES_RESOURCE, "mroldapi/pattern-format-{@timestamp:YYYY-MM-dd}-with-id");
Expand Down
7 changes: 7 additions & 0 deletions mr/src/main/java/org/elasticsearch/hadoop/cfg/ConfigurationOptions.java
View file
Expand Up @@ -152,6 +152,13 @@ public interface ConfigurationOptions {
String ES_MAPPING_CONSTANT_AUTO_QUOTE = "es.mapping.constant.auto.quote";
String ES_MAPPING_CONSTANT_AUTO_QUOTE_DEFAULT = "true";

String ES_MAPPING_VERSION_TYPE = "es.mapping.version.type";
String ES_MAPPING_VERSION_TYPE_INTERNAL = "internal";
String ES_MAPPING_VERSION_TYPE_EXTERNAL = "external";
String ES_MAPPING_VERSION_TYPE_EXTERNAL_GT = "external_gt";
String ES_MAPPING_VERSION_TYPE_EXTERNAL_GTE = "external_gte";
String ES_MAPPING_VERSION_TYPE_FORCE = "force";


/** Read settings */
String ES_READ_METADATA = "es.read.metadata";
Expand Down
42 changes: 28 additions & 14 deletions mr/src/main/java/org/elasticsearch/hadoop/cfg/Settings.java
View file
Expand Up @@ -40,7 +40,7 @@
*/
public abstract class Settings {

public String getNodes() {
public String getNodes() {
String host = getProperty(ES_HOST);
if (StringUtils.hasText(host)) {
throw new EsHadoopIllegalArgumentException(String.format("`%s` property has been deprecated - use `%s` instead", ES_HOST, ES_NODES));
Expand Down Expand Up @@ -106,9 +106,9 @@ public String getScrollFields() {
return (StringUtils.hasText(internalFields) ? internalFields : getProperty(ES_SCROLL_FIELDS));
}

public boolean getScrollEscapeUri() {
return Booleans.parseBoolean(getProperty(ES_SCROLL_ESCAPE_QUERY_URI, ES_SCROLL_ESCAPE_QUERY_URI_DEFAULT));
}
public boolean getScrollEscapeUri() {
return Booleans.parseBoolean(getProperty(ES_SCROLL_ESCAPE_QUERY_URI, ES_SCROLL_ESCAPE_QUERY_URI_DEFAULT));
}

public String getSerializerValueWriterClassName() {
return getProperty(ES_SERIALIZATION_WRITER_VALUE_CLASS);
Expand All @@ -134,9 +134,9 @@ public boolean getInputAsJson() {
return Booleans.parseBoolean(getProperty(ES_INPUT_JSON, ES_INPUT_JSON_DEFAULT));
}

public boolean getOutputAsJson() {
return Booleans.parseBoolean(getProperty(ES_OUTPUT_JSON, ES_OUTPUT_JSON_DEFAULT));
}
public boolean getOutputAsJson() {
return Booleans.parseBoolean(getProperty(ES_OUTPUT_JSON, ES_OUTPUT_JSON_DEFAULT));
}

public String getOperation() {
return getProperty(ES_WRITE_OPERATION, ES_WRITE_OPERATION_DEFAULT).toLowerCase(Locale.ENGLISH);
Expand All @@ -154,6 +154,20 @@ public String getMappingVersion() {
return getProperty(ES_MAPPING_VERSION);
}

public boolean hasMappingVersionType() {
String versionType = getMappingVersionType();
return (StringUtils.hasText(getMappingVersion()) && StringUtils.hasText(versionType) && !versionType.equals(ES_MAPPING_VERSION_TYPE_INTERNAL));
}

public String getMappingVersionType() {
String versionType = getProperty(ES_MAPPING_VERSION_TYPE);
// if no version type is specified, fall to defaults
if (!StringUtils.hasText(versionType)) {
return (StringUtils.hasText(getMappingVersion()) ? ES_MAPPING_VERSION_TYPE_EXTERNAL : "");
}
return versionType;
}

public String getMappingRouting() {
return getProperty(ES_MAPPING_ROUTING);
}
Expand Down Expand Up @@ -206,9 +220,9 @@ public String getMappingParamsExtractorClassName() {
return getProperty(ES_MAPPING_PARAMS_EXTRACTOR_CLASS, ES_MAPPING_PARAMS_DEFAULT_EXTRACTOR_CLASS);
}

public boolean getMappingConstantAutoQuote() {
return Booleans.parseBoolean(getProperty(ES_MAPPING_CONSTANT_AUTO_QUOTE, ES_MAPPING_CONSTANT_AUTO_QUOTE_DEFAULT));
}
public boolean getMappingConstantAutoQuote() {
return Booleans.parseBoolean(getProperty(ES_MAPPING_CONSTANT_AUTO_QUOTE, ES_MAPPING_CONSTANT_AUTO_QUOTE_DEFAULT));
}

public int getUpdateRetryOnConflict() {
return Integer.parseInt(getProperty(ES_UPDATE_RETRY_ON_CONFLICT, ES_UPDATE_RETRY_ON_CONFLICT_DEFAULT));
Expand Down Expand Up @@ -264,12 +278,12 @@ public String getNetworkSSLKeyStoreLocation() {
return getProperty(ES_NET_SSL_KEYSTORE_LOCATION);
}

public String getNetworkSSLProtocol() {
return getProperty(ES_NET_SSL_PROTOCOL, ES_NET_SSL_PROTOCOL_DEFAULT);
}
public String getNetworkSSLProtocol() {
return getProperty(ES_NET_SSL_PROTOCOL, ES_NET_SSL_PROTOCOL_DEFAULT);
}

public String getNetworkSSLKeyStoreType() {
return getProperty(ES_NET_SSL_KEYSTORE_TYPE, ES_NET_SSL_KEYSTORE_TYPE_DEFAULT);
return getProperty(ES_NET_SSL_KEYSTORE_TYPE, ES_NET_SSL_KEYSTORE_TYPE_DEFAULT);
}

public String getNetworkSSLKeyStorePass() {
Expand Down
5 changes: 5 additions & 0 deletions mr/src/main/java/org/elasticsearch/hadoop/serialization/bulk/AbstractBulkFactory.java
View file
Expand Up @@ -384,6 +384,11 @@ protected boolean ttl(List<Object> pieces) {

protected boolean version(List<Object> pieces) {
if (version() != null) {
if (settings.hasMappingVersionType()) {
pieces.add("\"_version_type\":\"");
pieces.add(settings.getMappingVersionType());
pieces.add("\"");
}
pieces.add("\"_version\":");
pieces.add(version());
return true;
Expand Down

0 comments on commit a5148b8

Please sign in to comment.