From ca449c0641a69a842b4026a8db31236ad1d94bb9 Mon Sep 17 00:00:00 2001
From: Oleksii Moskalenko <moskalenko.alexey@gmail.com>
Date: Fri, 8 Jan 2021 17:13:15 +0800
Subject: [PATCH] mapFieldWithParent

Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com>
---
 .../scala/feast/ingestion/IngestionJob.scala  | 14 ++++--
 .../feast/ingestion/utils/JsonUtils.scala     | 47 +++++++++++++++++++
 2 files changed, 58 insertions(+), 3 deletions(-)
 create mode 100644 spark/ingestion/src/main/scala/feast/ingestion/utils/JsonUtils.scala

diff --git a/spark/ingestion/src/main/scala/feast/ingestion/IngestionJob.scala b/spark/ingestion/src/main/scala/feast/ingestion/IngestionJob.scala
index 376d94ee019..86af18d07e6 100644
--- a/spark/ingestion/src/main/scala/feast/ingestion/IngestionJob.scala
+++ b/spark/ingestion/src/main/scala/feast/ingestion/IngestionJob.scala
@@ -16,6 +16,7 @@
  */
 package feast.ingestion
 
+import feast.ingestion.utils.JsonUtils
 import org.joda.time.DateTime
 import org.json4s._
 import org.json4s.jackson.JsonMethods.{parse => parseJSON}
@@ -38,13 +39,20 @@ object IngestionJob {
       .text("Mode to operate ingestion job (offline or online)")
 
     opt[String](name = "source")
-      .action((x, c) =>
-        parseJSON(x).camelizeKeys.extract[Sources] match {
+      .action((x, c) => {
+        val json = parseJSON(x)
+        JsonUtils
+          .mapFieldWithParent(json) {
+            case (parent: String, (key: String, v: JValue)) if !parent.equals("field_mapping") =>
+              JsonUtils.camelize(key) -> v
+            case (_, x) => x
+          }
+          .extract[Sources] match {
           case Sources(file: Some[FileSource], _, _)   => c.copy(source = file.get)
           case Sources(_, bq: Some[BQSource], _)       => c.copy(source = bq.get)
           case Sources(_, _, kafka: Some[KafkaSource]) => c.copy(source = kafka.get)
         }
-      )
+      })
       .required()
       .text("JSON-encoded source object (e.g. {\"kafka\":{\"bootstrapServers\":...}}")
 
diff --git a/spark/ingestion/src/main/scala/feast/ingestion/utils/JsonUtils.scala b/spark/ingestion/src/main/scala/feast/ingestion/utils/JsonUtils.scala
new file mode 100644
index 00000000000..6eeedad508f
--- /dev/null
+++ b/spark/ingestion/src/main/scala/feast/ingestion/utils/JsonUtils.scala
@@ -0,0 +1,47 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2021 The Feast Authors
+ *
+ * Licensed 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
+ *
+ *     https://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 feast.ingestion.utils
+
+import java.util.Locale.ENGLISH
+
+import org.json4s.{JArray, JField, JObject, JValue}
+
+object JsonUtils {
+  def mapFieldWithParent(jv: JValue)(f: (String, JField) => JField): JValue = {
+    def rec(v: JValue, parent: String = ""): JValue = v match {
+      case JObject(l) => JObject(l.map { case (key, va) => f(parent, key -> rec(va, key)) })
+      case JArray(l)  => JArray(l.map(rec(_, parent)))
+      case x          => x
+    }
+    rec(jv)
+  }
+
+  def camelize(word: String): String = {
+    if (word.nonEmpty) {
+      val w = pascalize(word)
+      w.substring(0, 1).toLowerCase(ENGLISH) + w.substring(1)
+    } else {
+      word
+    }
+  }
+
+  def pascalize(word: String): String = {
+    val lst = word.split("_").toList
+    (lst.headOption.map(s => s.substring(0, 1).toUpperCase(ENGLISH) + s.substring(1)).get ::
+      lst.tail.map(s => s.substring(0, 1).toUpperCase + s.substring(1))).mkString("")
+  }
+}