Add from_json function to convert json into flowtide objects (#733)

* Add to_json function to convert an object to a json string

* add check that to_json must have 1 arg

* Add from_json function to convert json into flowtide objects

* add docs

* fix comment

* Add so binaryvalue can also be used in from_json

Author: Ulimo

docs/docs/expressions/scalarfunctions/string.md

@@ -249,6 +249,18 @@ Converts an object into json stored as a string.
 
 ### SQL Usage
 
-```
+```sql
 SELECT to_json(column1) ...
+```
+
+## From Json
+
+This function does not have a substrait definition.
+
+Converts a JSON string to flowtide data objects. It is also possible to use a binary value in utf8 encoding as the input.
+
+### SQL Usage
+
+```sql
+SELECT from_json(myjsoncolumn) ...
 ```

src/FlowtideDotNet.Core/ColumnStore/Json/DataValueJsonReader.cs

@@ -0,0 +1,133 @@
+// 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
+//
+//     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.
+
+using FlowtideDotNet.Core.ColumnStore.DataValues;
+using System.Text.Json;
+
+namespace FlowtideDotNet.Core.ColumnStore.Json
+{
+    internal static class DataValueJsonReader
+    {
+        public static IDataValue Read(ref Utf8JsonReader reader)
+        {
+            while (reader.TokenType == JsonTokenType.Comment || reader.TokenType == JsonTokenType.None)
+            {
+                if (!reader.Read())
+                {
+                    throw new JsonException("Unexpected json");
+                }
+                continue;
+            }
+            switch (reader.TokenType)
+            {
+                case JsonTokenType.String:
+                    var strVal = reader.GetString();
+                    if (strVal == null)
+                    {
+                        return NullValue.Instance;
+                    }
+                    return new StringValue(strVal);
+                case JsonTokenType.False:
+                    return new BoolValue(false);
+                case JsonTokenType.True:
+                    return new BoolValue(true);
+                case JsonTokenType.Number:
+                    if (reader.TryGetInt64(out var intVal))
+                    {
+                        return new Int64Value(intVal);
+                    }
+                    if (reader.TryGetDouble(out var doubleVal))
+                    {
+                        return new DoubleValue(doubleVal);
+                    }
+                    throw new JsonException();
+                case JsonTokenType.Null:
+                    return NullValue.Instance;
+                case JsonTokenType.StartObject:
+                    return ParseObject(ref reader);
+                case JsonTokenType.StartArray:
+                    return ParseArray(ref reader);
+            }
+            throw new JsonException($"Unknown json token type {reader.TokenType}");
+        }
+
+        private static IDataValue ParseArray(ref Utf8JsonReader reader)
+        {
+            if (!reader.Read())
+            {
+                throw new JsonException("Unexpected json");
+            }
+
+            List<IDataValue> values = new List<IDataValue>();
+            while (true)
+            {
+                if (reader.TokenType == JsonTokenType.EndArray)
+                {
+                    break;
+                }
+
+                values.Add(Read(ref reader));
+
+                if (!reader.Read())
+                {
+                    throw new JsonException("Unexpected json");
+                }
+            }
+
+            return new ListValue(values);
+        }
+
+        private static IDataValue ParseObject(ref Utf8JsonReader reader)
+        {
+            if (!reader.Read())
+            {
+                throw new JsonException("Unexpected json");
+            }
+
+            List<KeyValuePair<IDataValue, IDataValue>> properties = new List<KeyValuePair<IDataValue, IDataValue>>();
+            while (true)
+            {
+                if (reader.TokenType == JsonTokenType.EndObject)
+                {
+                    break;
+                }
+                if (reader.TokenType != JsonTokenType.PropertyName)
+                {
+                    throw new JsonException();
+                }
+
+                var propertyName = reader.GetString();
+
+                if (propertyName == null)
+                {
+                    throw new JsonException("Unexpected json");
+                }
+
+                if (!reader.Read())
+                {
+                    throw new JsonException("Unexpected json");
+                }
+
+                var value = Read(ref reader);
+
+                if (!reader.Read())
+                {
+                    throw new JsonException("Unexpected json");
+                }
+
+                properties.Add(new KeyValuePair<IDataValue, IDataValue>(new StringValue(propertyName), value));
+            }
+
+            return new MapValue(properties);
+        }
+    }
+}

src/FlowtideDotNet.Core/Compute/Columnar/Functions/BuiltInStringFunctions.cs

@@ -23,6 +23,7 @@
 using System.Linq.Expressions;
 using System.Reflection;
 using System.Text;
+using System.Text.Json;
 using System.Text.RegularExpressions;
 using static SqlParser.Ast.MatchRecognizeSymbol;
 using static SqlParser.Ast.Partition;
@@ -202,6 +203,8 @@ public static void RegisterFunctions(IFunctionsRegister functionsRegister)
                 var resultContainer = Expression.Constant(new DataValueContainer());
                 return System.Linq.Expressions.Expression.Call(genericMethod, expr, resultContainer, jsonWriterConstant);
             });
+
+            functionsRegister.RegisterScalarMethod(FunctionsString.Uri, FunctionsString.FromJson, typeof(BuiltInStringFunctions), nameof(FromJsonImplementation));
         }
 
         private static bool SubstringTryGetParameters<T1, T2, T3>(
@@ -804,5 +807,65 @@ private static DataValueContainer ToJsonImplementation<T1>(in T1 val, DataValueC
             result._stringValue = new StringValue(writer.WrittenMemory);
             return result;
         }
+
+        private static IDataValue FromJsonImplementation<T1>(in T1 val)
+            where T1 : IDataValue
+        {
+            if (val.Type != ArrowTypeId.String && val.Type != ArrowTypeId.Binary)
+            {
+                return NullValue.Instance;
+            }
+
+            if (val.Type == ArrowTypeId.String)
+            {
+                Utf8JsonReader reader = new Utf8JsonReader(val.AsString.Span);
+                try
+                {
+                    return DataValueJsonReader.Read(ref reader);
+                }
+                catch (JsonException)
+                {
+                    return NullValue.Instance;
+                }
+            }
+            else
+            {
+                Utf8JsonReader reader = new Utf8JsonReader(val.AsBinary);
+                try
+                {
+                    return DataValueJsonReader.Read(ref reader);
+                }
+                catch (JsonException)
+                {
+                    return NullValue.Instance;
+                }
+            }
+
+               
+        }
+
+        private static IDataValue FromJsonImplementation_error_handling__ERROR<T1>(in T1 val)
+            where T1 : IDataValue
+        {
+            if (val.Type == ArrowTypeId.Null)
+            {
+                return NullValue.Instance;
+            }
+            if (val.Type != ArrowTypeId.String && val.Type != ArrowTypeId.Binary)
+            {
+                throw new ArgumentException("FromJson function must have a string or binary argument.");
+            }
+
+            if (val.Type == ArrowTypeId.String)
+            {
+                Utf8JsonReader reader = new Utf8JsonReader(val.AsString.Span);
+                return DataValueJsonReader.Read(ref reader);
+            }
+            else
+            {
+                Utf8JsonReader reader = new Utf8JsonReader(val.AsBinary);
+                return DataValueJsonReader.Read(ref reader);
+            }
+        }
     }
 }

src/FlowtideDotNet.Substrait/FunctionExtensions/FunctionsString.cs

@@ -33,6 +33,7 @@ public static class FunctionsString
         public const string StringSplit = "string_split";
         public const string RegexStringSplit = "regexp_string_split";
         public const string ToJson = "to_json";
+        public const string FromJson = "from_json";
 
         public const string StringAgg = "string_agg";
     }

src/FlowtideDotNet.Substrait/Sql/Internal/BuiltInSqlFunctions.cs

@@ -755,6 +755,7 @@ public static void AddBuiltInFunctions(SqlFunctionRegister sqlFunctionRegister)
             RegisterTwoVariableScalarFunction(sqlFunctionRegister, "regexp_string_split", FunctionsString.Uri, FunctionsString.RegexStringSplit);
 
             RegisterOneVariableScalarFunction(sqlFunctionRegister, "to_json", FunctionsString.Uri, FunctionsString.ToJson);
+            RegisterOneVariableScalarFunction(sqlFunctionRegister, "from_json", FunctionsString.Uri, FunctionsString.FromJson);
 
             // Table functions
             UnnestSqlFunction.AddUnnest(sqlFunctionRegister);

tests/FlowtideDotNet.AcceptanceTests/StringFunctionTests.cs

@@ -229,5 +229,16 @@ INSERT INTO output
             await WaitForUpdate();
             AssertCurrentDataEqual(Users.Select(x => new { json = JsonSerializer.Serialize(new {firstName = x.FirstName, lastName = x.LastName}) }));
         }
+
+        [Fact]
+        public async Task FromJsonWithMap()
+        {
+            GenerateData();
+            await StartStream(@"
+            INSERT INTO output 
+            SELECT from_json(to_json(map('firstName', firstName, 'lastName', lastName))) as json FROM users");
+            await WaitForUpdate();
+            AssertCurrentDataEqual(Users.Select(x => new { json = new { firstName = x.FirstName, lastName = x.LastName } }));
+        }
     }
 }

tests/FlowtideDotNet.ComputeTests/Internal/Framework/SubstraitTestRunner.cs

@@ -127,15 +127,32 @@ private ValueTask<TimeSpan> InvokeScalarTest(SubstraitTestRunnerContext ctxt, ob
             }
             Column resultColumn = Column.Create(GlobalMemoryManager.Instance);
             // Run once to try and reduce IL compile time for output
-            compiledMethod(new EventBatchData(columns), 0, resultColumn);
+            if (parsedTest.Expected.ExpectError)
+            {
+                Assert.ThrowsAny<Exception>(() => compiledMethod(new EventBatchData(columns), 0, resultColumn));
+            }
+            else
+            {
+                compiledMethod(new EventBatchData(columns), 0, resultColumn);
+            }
             resultColumn.Clear();
             Stopwatch sw = new();
             sw.Start();
-            compiledMethod(new EventBatchData(columns), 0, resultColumn);
+            if (parsedTest.Expected.ExpectError)
+            {
+                Assert.ThrowsAny<Exception>(() => compiledMethod(new EventBatchData(columns), 0, resultColumn));
+            }
+            else
+            {
+                compiledMethod(new EventBatchData(columns), 0, resultColumn);
+            }
             sw.Stop();
 
-            var actual = resultColumn.GetValueAt(0, default);
-            Assert.Equal(parsedTest.Expected.ExpectedValue, actual, (x, y) => DataValueComparer.CompareTo(x!, y!) == 0);
+            if (!parsedTest.Expected.ExpectError)
+            {
+                var actual = resultColumn.GetValueAt(0, default);
+                Assert.Equal(parsedTest.Expected.ExpectedValue, actual, (x, y) => DataValueComparer.CompareTo(x!, y!) == 0);
+            }
             return new(sw.Elapsed);
         }
 

tests/FlowtideDotNet.ComputeTests/Internal/Parser/Tests/ArgumentVisitor.cs

@@ -15,6 +15,7 @@
 using FlowtideDotNet.Core.ColumnStore;
 using FlowtideDotNet.Core.ColumnStore.DataValues;
 using System.Globalization;
+using System.Text;
 
 namespace FlowtideDotNet.ComputeTests.Internal.Tests
 {
@@ -49,7 +50,11 @@ public override IDataValue VisitIntegerLiteral([NotNull] FuncTestCaseParser.Inte
 
         public override IDataValue VisitFixedBinaryArg([NotNull] FuncTestCaseParser.FixedBinaryArgContext context)
         {
-            throw new NotImplementedException();
+            var text = context.GetText();
+
+            text = text.Trim('\'');
+
+            return new BinaryValue(Encoding.UTF8.GetBytes(text));
         }
 
         public override IDataValue VisitStringArg([NotNull] FuncTestCaseParser.StringArgContext context)

tests/FlowtideDotNet.ComputeTests/OtherCases/string/from_json.test

@@ -0,0 +1,29 @@
+### SUBSTRAIT_SCALAR_TEST: v1.0
+### SUBSTRAIT_INCLUDE: '/extensions/functions_string.yaml'
+
+# basic: Basic examples without any special cases
+from_json('"abcdef"'::str) = 'abcdef'::str
+from_json('["a", "b"]'::str) = ['a', 'b']::LIST<str>
+from_json('123'::str) = 123::i64
+from_json('123.45'::str) = 123.45::fp64
+from_json('true'::str) = true::bool
+from_json('false'::str) = false::bool
+from_json('null'::str) = null::str
+
+# Binary input: example with binary input
+from_json('"abcdef"'::FIXEDBINARY<7>) = 'abcdef'::str
+from_json('["a", "b"]'::FIXEDBINARY<10>) = ['a', 'b']::LIST<str>
+
+# Errors: error handling for invalid json
+from_json('{"a": 1'::str) = null::str
+from_json(123::i64) = null::str
+from_json('{"a": 1'::FIXEDBINARY<7>) = null::str
+from_json('{"a": 1'::str) [error_handling:ERROR] = <!ERROR>
+from_json(123::i64) [error_handling:ERROR] = <!ERROR>
+from_json('{"a": 1'::FIXEDBINARY<7>) [error_handling:ERROR] = <!ERROR>
+
+# Errors: Error handling enabled with correct json
+from_json('"abcdef"'::str) [error_handling:ERROR] = 'abcdef'::str
+from_json('["a", "b"]'::str) [error_handling:ERROR] = ['a', 'b']::LIST<str>
+from_json('"abcdef"'::FIXEDBINARY<7>) [error_handling:ERROR] = 'abcdef'::str
+from_json('["a", "b"]'::FIXEDBINARY<10>) [error_handling:ERROR] = ['a', 'b']::LIST<str>