Support OpenSearch ip field type

Signed-off-by: Tomoyuki Morita <[email protected]>
opensearch-project · Feb 11, 2025 · 0174ca5 · 0174ca5
1 parent 1a6b9f3
commit 0174ca5
Show file tree

Hide file tree

Showing 4 changed files with 56 additions and 0 deletions.
diff --git a/docs/opensearch-table.md b/docs/opensearch-table.md
@@ -63,6 +63,7 @@ The following table defines the data type mapping between OpenSearch index field
 | date(Date)               | DateType                          |
 | keyword                  | StringType, VarcharType, CharType |
 | text                     | StringType(meta(osType)=text)     |
+| ip                       | StringType(meta(osType)=ip)       |
 | object                   | StructType                        |
 | alias                    | Inherits referenced field type    |
 

diff --git a/...-spark-integration/src/main/scala/org/apache/spark/sql/flint/datatype/FlintDataType.scala b/...-spark-integration/src/main/scala/org/apache/spark/sql/flint/datatype/FlintDataType.scala
@@ -106,6 +106,11 @@ object FlintDataType {
       // binary types
       case JString("binary") => BinaryType
 
+      // ip type
+      case JString("ip") =>
+        metadataBuilder.putString("osType", "ip")
+        StringType
+
       // not supported
       case unknown => throw new IllegalStateException(s"unsupported data type: $unknown")
     }

diff --git a/...src/integration/scala/org/apache/spark/opensearch/table/OpenSearchTableQueryITSuite.scala b/...src/integration/scala/org/apache/spark/opensearch/table/OpenSearchTableQueryITSuite.scala
@@ -65,4 +65,27 @@ class OpenSearchTableQueryITSuite extends OpenSearchCatalogSuite with FlintPPLSu
       checkAnswer(df, Row(1, 1))
     }
   }
+
+  test("Query index with ip data type") {
+    val index1 = "t0001"
+    withIndexName(index1) {
+      indexWithIp(index1)
+
+      var df = spark.sql(s"""SELECT client, server FROM ${catalogName}.default.$index1""")
+      val ip0 = "192.168.0.10"
+      val ip1 = "192.168.0.11"
+      val ip2 = "100.10.12.123"
+      val ip3 = "2001:db8:3333:4444:5555:6666:7777:8888"
+      val ip4 = "2001:db8::1234:5678"
+      checkAnswer(df, Seq(Row(ip0, ip2), Row(ip1, ip2), Row(ip3, ip4)))
+
+      df = spark.sql(
+        s"""SELECT client, server FROM ${catalogName}.default.$index1 WHERE client = '192.168.0.10'""")
+      checkAnswer(df, Seq(Row(ip0, ip2)))
+
+      df = spark.sql(
+        s"""SELECT client, server FROM ${catalogName}.default.$index1 WHERE server = '2001:db8::1234:5678'""")
+      checkAnswer(df, Seq(Row(ip3, ip4)))
+    }
+  }
 }
diff --git a/integ-test/src/integration/scala/org/opensearch/flint/OpenSearchSuite.scala b/integ-test/src/integration/scala/org/opensearch/flint/OpenSearchSuite.scala
@@ -139,6 +139,33 @@ trait OpenSearchSuite extends BeforeAndAfterAll {
     index(indexName, oneNodeSetting, mappings, docs)
   }
 
+  def indexWithIp(indexName: String): Unit = {
+    val mappings = """{
+                     |  "properties": {
+                     |    "client": {
+                     |      "type": "ip"
+                     |    },
+                     |    "server": {
+                     |      "type": "ip"
+                     |    }
+                     |  }
+                     |}""".stripMargin
+    val docs = Seq(
+      """{
+                     |  "client": "192.168.0.10",
+                     |  "server": "100.10.12.123"
+                     |}""".stripMargin,
+      """{
+                    |  "client": "192.168.0.11",
+                    |  "server": "100.10.12.123"
+                    |}""".stripMargin,
+      """{
+                    |  "client": "2001:db8:3333:4444:5555:6666:7777:8888",
+                    |  "server": "2001:db8::1234:5678"
+                    |}""".stripMargin)
+    index(indexName, oneNodeSetting, mappings, docs)
+  }
+
   def index(index: String, settings: String, mappings: String, docs: Seq[String]): Unit = {
     openSearchClient.indices.create(
       new CreateIndexRequest(index)