Adding support for wildcard fields (#1823)

masseyke · web-flow · commit 56a0d1367831 · 2022-01-06T16:19:17.000-06:00
Adding support for wildcard fields in Elasticsearch mappings.
diff --git a/mr/src/main/java/org/elasticsearch/hadoop/serialization/FieldType.java b/mr/src/main/java/org/elasticsearch/hadoop/serialization/FieldType.java
@@ -44,6 +44,7 @@ public enum FieldType {
     TOKEN_COUNT,
     // ES 5.x
     TEXT, KEYWORD, HALF_FLOAT, SCALED_FLOAT,
+    WILDCARD,
 
     GEO_POINT,
     GEO_SHAPE,
@@ -81,6 +82,7 @@ public enum FieldType {
         CAST_HIERARCHY.put(TOKEN_COUNT,      new LinkedHashSet<FieldType>(Arrays.asList(LONG, KEYWORD)));
         CAST_HIERARCHY.put(TEXT,             new LinkedHashSet<FieldType>(Collections.singletonList(KEYWORD)));
         CAST_HIERARCHY.put(KEYWORD,          new LinkedHashSet<FieldType>());
+        CAST_HIERARCHY.put(WILDCARD,         new LinkedHashSet<FieldType>(Collections.singletonList(KEYWORD)));
         CAST_HIERARCHY.put(HALF_FLOAT,       new LinkedHashSet<FieldType>(Arrays.asList(FLOAT, DOUBLE, KEYWORD)));
         CAST_HIERARCHY.put(SCALED_FLOAT,     new LinkedHashSet<FieldType>(Arrays.asList(DOUBLE, KEYWORD)));
         CAST_HIERARCHY.put(GEO_POINT,        new LinkedHashSet<FieldType>());
diff --git a/mr/src/test/java/org/elasticsearch/hadoop/serialization/dto/mapping/MappingTest.java b/mr/src/test/java/org/elasticsearch/hadoop/serialization/dto/mapping/MappingTest.java
@@ -54,6 +54,7 @@
 import static org.elasticsearch.hadoop.serialization.FieldType.SHORT;
 import static org.elasticsearch.hadoop.serialization.FieldType.STRING;
 import static org.elasticsearch.hadoop.serialization.FieldType.TEXT;
+import static org.elasticsearch.hadoop.serialization.FieldType.WILDCARD;
 
 import static org.elasticsearch.hadoop.serialization.dto.mapping.MappingUtils.findTypos;
 import static org.junit.Assert.assertEquals;
@@ -167,6 +168,8 @@ public void testPrimitivesParsing() throws Exception {
         assertEquals(SCALED_FLOAT, props[13].type());
         assertEquals("field15", props[14].name());
         assertEquals(DATE_NANOS, props[14].type());
+        assertEquals("field16", props[15].name());
+        assertEquals(WILDCARD, props[15].type());
     }
 
     @Test
diff --git a/mr/src/test/resources/org/elasticsearch/hadoop/serialization/dto/mapping/typed/primitives.json b/mr/src/test/resources/org/elasticsearch/hadoop/serialization/dto/mapping/typed/primitives.json
@@ -48,6 +48,9 @@
           },
           "field15" : {
             "type" : "date_nanos"
+          },
+          "field16" : {
+            "type" : "wildcard"
           }
         }
       }
diff --git a/spark/core/src/main/scala/org/elasticsearch/spark/serialization/ScalaValueReader.scala b/spark/core/src/main/scala/org/elasticsearch/spark/serialization/ScalaValueReader.scala
@@ -47,6 +47,7 @@ import org.elasticsearch.hadoop.serialization.FieldType.SHORT
 import org.elasticsearch.hadoop.serialization.FieldType.STRING
 import org.elasticsearch.hadoop.serialization.FieldType.TEXT
 import org.elasticsearch.hadoop.serialization.FieldType.TOKEN_COUNT
+import org.elasticsearch.hadoop.serialization.FieldType.WILDCARD
 import org.elasticsearch.hadoop.serialization.Parser
 import org.elasticsearch.hadoop.serialization.Parser.Token.VALUE_BOOLEAN
 import org.elasticsearch.hadoop.serialization.Parser.Token.VALUE_NULL
@@ -82,6 +83,7 @@ class ScalaValueReader extends AbstractValueReader with SettingsAware {
         case STRING => textValue(value, parser)
         case TEXT => textValue(value, parser)
         case KEYWORD => textValue(value, parser)
+        case WILDCARD => textValue(value, parser)
         case BYTE => byteValue(value, parser)
         case SHORT => shortValue(value, parser)
         case INTEGER => intValue(value, parser)
diff --git a/spark/sql-13/src/itest/scala/org/elasticsearch/spark/integration/AbstractScalaEsSparkSQL.scala b/spark/sql-13/src/itest/scala/org/elasticsearch/spark/integration/AbstractScalaEsSparkSQL.scala
@@ -2286,6 +2286,32 @@ class AbstractScalaEsScalaSparkSQL(prefix: String, readMetadata: jl.Boolean, pus
     assertEquals(0, result(0).size)
   }
 
+  @Test
+  def testWildcard() {
+    val mapping = wrapMapping("data", s"""{
+                                         |      "properties": {
+                                         |        "name": {
+                                         |          "type": "wildcard"
+                                         |        }
+                                         |      }
+                                         |  }
+    """.stripMargin)
+
+    val index = wrapIndex("sparksql-test-wildcard")
+    val typed = "data"
+    val (target, docPath) = makeTargets(index, typed)
+    RestUtils.touch(index)
+    RestUtils.putMapping(index, typed, mapping.getBytes(StringUtils.UTF_8))
+    val wildcardDocument = """{ "name" : "Chipotle Mexican Grill"}""".stripMargin
+    sc.makeRDD(Seq(wildcardDocument)).saveJsonToEs(target)
+    RestUtils.refresh(index)
+    val df = sqc.read.format("es").load(index)
+    val dataType = df.schema("name").dataType
+    assertEquals("string", dataType.typeName)
+    val head = df.head()
+    assertThat(head.getString(0), containsString("Chipotle"))
+  }
+
   /**
    * Take advantage of the fixed method order and clear out all created indices.
    * The indices will last in Elasticsearch for all parameters of this test suite.
diff --git a/spark/sql-13/src/main/scala/org/elasticsearch/spark/sql/SchemaUtils.scala b/spark/sql-13/src/main/scala/org/elasticsearch/spark/sql/SchemaUtils.scala
@@ -68,6 +68,7 @@ import org.elasticsearch.hadoop.serialization.FieldType.OBJECT
 import org.elasticsearch.hadoop.serialization.FieldType.SHORT
 import org.elasticsearch.hadoop.serialization.FieldType.STRING
 import org.elasticsearch.hadoop.serialization.FieldType.TEXT
+import org.elasticsearch.hadoop.serialization.FieldType.WILDCARD
 import org.elasticsearch.hadoop.serialization.dto.mapping.Field
 import org.elasticsearch.hadoop.serialization.dto.mapping.GeoField
 import org.elasticsearch.hadoop.serialization.dto.mapping.GeoPointType
@@ -163,6 +164,7 @@ private[sql] object SchemaUtils {
       case STRING    => StringType
       case TEXT      => StringType
       case KEYWORD   => StringType
+      case WILDCARD  => StringType
       case HALF_FLOAT => FloatType
       case SCALED_FLOAT => FloatType
       case DATE      => if (cfg.getMappingDateRich) TimestampType else StringType
diff --git a/spark/sql-20/src/itest/scala/org/elasticsearch/spark/integration/AbstractScalaEsSparkSQL.scala b/spark/sql-20/src/itest/scala/org/elasticsearch/spark/integration/AbstractScalaEsSparkSQL.scala
@@ -2368,6 +2368,32 @@ class AbstractScalaEsScalaSparkSQL(prefix: String, readMetadata: jl.Boolean, pus
     assertEquals(0, result(0).size)
   }
 
+  @Test
+  def testWildcard() {
+    val mapping = wrapMapping("data", s"""{
+                                         |      "properties": {
+                                         |        "name": {
+                                         |          "type": "wildcard"
+                                         |        }
+                                         |      }
+                                         |  }
+    """.stripMargin)
+
+    val index = wrapIndex("sparksql-test-wildcard")
+    val typed = "data"
+    val (target, docPath) = makeTargets(index, typed)
+    RestUtils.touch(index)
+    RestUtils.putMapping(index, typed, mapping.getBytes(StringUtils.UTF_8))
+    val wildcardDocument = """{ "name" : "Chipotle Mexican Grill"}""".stripMargin
+    sc.makeRDD(Seq(wildcardDocument)).saveJsonToEs(target)
+    RestUtils.refresh(index)
+    val df = sqc.read.format("es").load(index)
+    val dataType = df.schema("name").dataType
+    assertEquals("string", dataType.typeName)
+    val head = df.head()
+    assertThat(head.getString(0), containsString("Chipotle"))
+  }
+
   /**
    * Take advantage of the fixed method order and clear out all created indices.
    * The indices will last in Elasticsearch for all parameters of this test suite.
diff --git a/spark/sql-20/src/main/scala/org/elasticsearch/spark/sql/SchemaUtils.scala b/spark/sql-20/src/main/scala/org/elasticsearch/spark/sql/SchemaUtils.scala
@@ -68,6 +68,7 @@ import org.elasticsearch.hadoop.serialization.FieldType.OBJECT
 import org.elasticsearch.hadoop.serialization.FieldType.SHORT
 import org.elasticsearch.hadoop.serialization.FieldType.STRING
 import org.elasticsearch.hadoop.serialization.FieldType.TEXT
+import org.elasticsearch.hadoop.serialization.FieldType.WILDCARD
 import org.elasticsearch.hadoop.serialization.dto.mapping.Field
 import org.elasticsearch.hadoop.serialization.dto.mapping.GeoField
 import org.elasticsearch.hadoop.serialization.dto.mapping.GeoPointType
@@ -165,6 +166,7 @@ private[sql] object SchemaUtils {
       case STRING       => StringType
       case TEXT         => StringType
       case KEYWORD      => StringType
+      case WILDCARD      => StringType
       case DATE         => if (cfg.getMappingDateRich) TimestampType else StringType
       case DATE_NANOS => if (cfg.getMappingDateRich) TimestampType else StringType
       case OBJECT       => convertToStruct(field, geoInfo, absoluteName, arrayIncludes, arrayExcludes, cfg)
diff --git a/spark/sql-30/src/itest/scala/org/elasticsearch/spark/integration/AbstractScalaEsSparkSQL.scala b/spark/sql-30/src/itest/scala/org/elasticsearch/spark/integration/AbstractScalaEsSparkSQL.scala
@@ -2369,6 +2369,31 @@ class AbstractScalaEsScalaSparkSQL(prefix: String, readMetadata: jl.Boolean, pus
     assertEquals(0, result(0).size)
   }
 
+  @Test
+  def testWildcard() {
+    val mapping = wrapMapping("data", s"""{
+                                         |      "properties": {
+                                         |        "name": {
+                                         |          "type": "wildcard"
+                                         |        }
+                                         |      }
+                                         |  }
+    """.stripMargin)
+
+    val index = wrapIndex("sparksql-test-wildcard")
+    val typed = "data"
+    val (target, docPath) = makeTargets(index, typed)
+    RestUtils.touch(index)
+    RestUtils.putMapping(index, typed, mapping.getBytes(StringUtils.UTF_8))
+    val wildcardDocument = """{ "name" : "Chipotle Mexican Grill"}""".stripMargin
+    sc.makeRDD(Seq(wildcardDocument)).saveJsonToEs(target)
+    RestUtils.refresh(index)
+    val df = sqc.read.format("es").load(index)
+    val dataType = df.schema("name").dataType
+    assertEquals("string", dataType.typeName)
+    val head = df.head()
+    assertThat(head.getString(0), containsString("Chipotle"))
+  }
 
   /**
    * Take advantage of the fixed method order and clear out all created indices.
diff --git a/spark/sql-30/src/main/scala/org/elasticsearch/spark/sql/SchemaUtils.scala b/spark/sql-30/src/main/scala/org/elasticsearch/spark/sql/SchemaUtils.scala
@@ -69,6 +69,7 @@ import org.elasticsearch.hadoop.serialization.FieldType.OBJECT
 import org.elasticsearch.hadoop.serialization.FieldType.SHORT
 import org.elasticsearch.hadoop.serialization.FieldType.STRING
 import org.elasticsearch.hadoop.serialization.FieldType.TEXT
+import org.elasticsearch.hadoop.serialization.FieldType.WILDCARD
 import org.elasticsearch.hadoop.serialization.dto.mapping.Field
 import org.elasticsearch.hadoop.serialization.dto.mapping.GeoField
 import org.elasticsearch.hadoop.serialization.dto.mapping.GeoPointType
@@ -165,6 +166,7 @@ private[sql] object SchemaUtils {
       case STRING       => StringType
       case TEXT         => StringType
       case KEYWORD      => StringType
+      case WILDCARD     => StringType
       case DATE         => if (cfg.getMappingDateRich) TimestampType else StringType
       case DATE_NANOS   => if (cfg.getMappingDateRich) TimestampType else StringType
       case OBJECT       => convertToStruct(field, geoInfo, absoluteName, arrayIncludes, arrayExcludes, cfg)

Original file line number	Diff line number	Diff line change
`@@ -48,6 +48,9 @@`
`48`	`48`	`},`
`49`	`49`	`"field15" : {`
`50`	`50`	`"type" : "date_nanos"`
	`51`	`+ },`
	`52`	`+ "field16" : {`
	`53`	`+ "type" : "wildcard"`
`51`	`54`	`}`
`52`	`55`	`}`
`53`	`56`	`}`