Skip to content

[SQL] JSON_* and VARIANT_* functions for filtering and transformations #6686

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mihaibudiu wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[SQL] JSON_* and VARIANT_* functions for filtering and transformations #6686

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
872 changes: 871 additions & 1 deletion crates/sqllib/src/variant.rs
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion docs.feldera.com/docs/sql/casts.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ SELECT cast(row(1, 2) as row(a integer, b tinyint)) as r;

## Safe casts

The `SAFE_CAST` function has the same syntax as `CAST`. `SAFE_CAST`
<a id="safe_cast"></a>The `SAFE_CAST` function has the same syntax as `CAST`. `SAFE_CAST`
produces the same result as `CAST` for all legal inputs. The main
difference is that `SAFE_CAST` never produces a runtime error,
producing a `NULL` value when a conversion is illegal.
Expand Down
15 changes: 14 additions & 1 deletion docs.feldera.com/docs/sql/function-index.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,14 @@
* `IS UNKNOWN`: [operators](operators.md#isnull)
* `IS_INF`: [float](float.md#is_inf)
* `IS_NAN`: [float](float.md#is_nan)
* `JSON_EACH_BIGINT`: [json](json.md#json_each)
* `JSON_EACH_BOOLEAN`: [json](json.md#json_each)
* `JSON_EACH_DATE`: [json](json.md#json_each)
* `JSON_EACH_STRING`: [json](json.md#json_each)
* `JSON_EACH_TIME`: [json](json.md#json_each)
* `JSON_EACH_TIMESTAMP`: [json](json.md#json_each)
* `JSON_KEYS`: [json](json.md#json_keys)
* `JSON_OBJECT_KEYS`: [json](json.md#json_object_keys)
* `LAG` (aggregate): [aggregates](aggregates.md#lag)
* `LATERAL`: [grammar](grammar.md#lateral)
* `LEAD` (aggregate): [aggregates](aggregates.md#lead)
Expand Down Expand Up @@ -212,7 +220,7 @@
* `ROUND`: [float](float.md#round), [float](float.md#round2), [decimal](decimal.md#round), [decimal](decimal.md#round2)
* `ROW`: [types](types.md#row_constructor)
* `ROW_NUMBER` (aggregate): [aggregates](aggregates.md#row_number)
* `SAFE_CAST`: [casts](casts.md#safe-casts)
* `SAFE_CAST`: [casts](casts.md#safe_cast)
* `SAFE_OFFSET`: [array](array.md#safe_offset)
* `SEC`: [float](float.md#sec)
* `SECH`: [float](float.md#sech)
Expand Down Expand Up @@ -258,6 +266,11 @@
* `VAR_POP` (aggregate): [aggregates](aggregates.md#var_pop)
* `VAR_SAMP` (aggregate): [aggregates](aggregates.md#var_samp)
* `VARIANCE` (aggregate): [aggregates](aggregates.md#variance)
* `VARIANT_DEEP_FILTER`: [json](json.md#variant_deep_filter)
* `VARIANT_DEEP_MAP`: [json](json.md#variant_deep_map)
* `VARIANT_FILTER`: [json](json.md#variant_filter)
* `VARIANT_MAP`: [json](json.md#variant_map)
* `VARIANT_MERGE`: [json](json.md#variant_merge)
* `WEEK`: [datetime](datetime.md#week)
* `XXHASH`: [string](string.md#xxhash), [binary](binary.md#xxhash)
* `YEAR`: [datetime](datetime.md#year)
Expand Down
324 changes: 324 additions & 0 deletions docs.feldera.com/docs/sql/json.md
View file
Open in desktop

Large diffs are not rendered by default.

41 changes: 40 additions & 1 deletion ...SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/ExpressionCompiler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1596,12 +1596,26 @@ else if (arg0Type.is(DBSPTypeMap.class))
case "typeof":
return compileFunction(call, node, type, ops, 1);
case "parse_json":
case "to_json": {
case "to_json":
case "json_each_bigint":
case "json_each_string":
case "json_each_boolean":
case "json_each_date":
case "json_each_time":
case "json_each_timestamp":
case "json_keys":
case "json_object_keys": {
DBSPExpression expr = this.strictnessCheck(ops, type);
if (expr != null)
return expr;
return compilePolymorphicFunction(false, call, node, type, ops, 1);
}
case "variant_merge": {
DBSPExpression expr = this.strictnessCheck(ops, type);
if (expr != null)
return expr;
return compilePolymorphicFunction(false, call, node, type, ops, 2);
}
case "sequence": {
for (int i = 0; i < ops.size(); i++)
this.ensureInteger(node, ops, i);
Expand Down Expand Up @@ -1721,6 +1735,31 @@ else if (arg0Type.is(DBSPTypeMap.class))
node, method, type.withMayBeNull(nullable), ops.get(0), ops.get(1))
.cast(node, type, DBSPCastExpression.CastType.SqlUnsafe);
}
case "variant_filter":
case "variant_deep_filter": {
validateArgCount(node, operationName, ops.size(), 2);
// The result is nullable regardless of the argument:
// a dropped non-map variant produces NULL
String method = opName +
ops.get(0).getType().nullableUnderlineSuffix();
return new DBSPApplyExpression(node, method, type, ops.get(0), ops.get(1));
}
case "variant_map":
case "variant_deep_map": {
validateArgCount(node, operationName, ops.size(), 2);
// The lambda may produce a value of any type;
// convert its result to a VARIANT
DBSPClosureExpression mapper = ops.get(1).to(DBSPClosureExpression.class);
DBSPType variant = DBSPTypeVariant.INSTANCE_NULLABLE;
DBSPExpression converted = mapper;
if (!mapper.getResultType().sameType(variant))
converted = mapper.body
.cast(node, variant, DBSPCastExpression.CastType.SqlUnsafe)
.closure(mapper.parameters);
String method = opName +
ops.get(0).getType().nullableUnderlineSuffix();
return new DBSPApplyExpression(node, method, type, ops.get(0), converted);
}
case "convert_timezone": {
validateArgCount(node, operationName, ops.size(), 3);
ops.set(0, this.makeTimezone(ops.get(0)));
Expand Down
9 changes: 8 additions & 1 deletion ...rc/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/CalciteFunctions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -711,8 +711,15 @@ public SqlOperatorTable getFunctions() {
if (func.library != SqlLibrary.STANDARD)
operators.add(func.function);
}
// Remove some standard functions that we don't want.
List<SqlOperator> standard = new ArrayList<>();
for (SqlOperator operator : SqlLibraryOperatorTableFactory.INSTANCE
.getOperatorTable(SqlLibrary.STANDARD).getOperatorList()) {
if (!operator.getName().startsWith("JSON_"))

@mythical-fred mythical-fred Jul 21, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This removes all standard Calcite JSON_* functions (JSON_VALUE, JSON_QUERY, JSON_EXISTS, JSON_ARRAY, JSON_OBJECT, JSON_DEPTH, JSON_LENGTH, JSON_PRETTY, JSON_REMOVE, etc.) — not just the ones that conflict with the new Feldera functions. The direct conflict is JSON_KEYS and JSON_OBJECT_KEYS, but the rest are collateral. Were any of the removed functions previously usable in Feldera SQL? If so, this is a breaking change worth documenting. If not (because Feldera uses VARIANT instead of string-based JSON), then this is fine — but a comment saying "Feldera replaces all standard JSON functions with VARIANT-based equivalents" would make the intent clearer for future readers.

standard.add(operator);
}
return SqlOperatorTables.chain(
SqlLibraryOperatorTableFactory.INSTANCE.getOperatorTable(SqlLibrary.STANDARD),
new SqlToRelCompiler.CaseInsensitiveOperatorTable(standard),
new SqlToRelCompiler.CaseInsensitiveOperatorTable(
SqlOperatorTables.spatialInstance().getOperatorList()),
new SqlToRelCompiler.CaseInsensitiveOperatorTable(operators));
Expand Down
89 changes: 89 additions & 0 deletions ...src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/CustomFunctions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler;

import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlCallBinding;
Expand Down Expand Up @@ -77,6 +78,14 @@ public CustomFunctions() {
this.functions.add(GreatestNonNullsFunction.INSTANCE);
this.functions.add(GunzipFunction.INSTANCE);
this.functions.add(InitcapSpacesFunction.INSTANCE);
this.functions.add(JsonEachFunction.BIGINT);
this.functions.add(JsonEachFunction.BOOLEAN);
this.functions.add(JsonEachFunction.DATE);
this.functions.add(JsonEachFunction.STRING);
this.functions.add(JsonEachFunction.TIME);
this.functions.add(JsonEachFunction.TIMESTAMP);
this.functions.add(JsonKeysFunction.KEYS);
this.functions.add(JsonKeysFunction.OBJECT_KEYS);
this.functions.add(LeastNonNullsFunction.INSTANCE);
this.functions.add(MakeDateFunction.INSTANCE);
this.functions.add(MakeTimeFunction.INSTANCE);
Expand All @@ -94,6 +103,11 @@ public CustomFunctions() {
this.functions.add(SplitPartFunction.INSTANCE);
this.functions.add(ToIntFunction.INSTANCE);
this.functions.add(ToJsonFunction.INSTANCE);
this.functions.add(VariantFilterFunction.DEEP);
this.functions.add(VariantFilterFunction.INSTANCE);
this.functions.add(VariantMapFunction.DEEP);
this.functions.add(VariantMapFunction.INSTANCE);
this.functions.add(VariantMergeFunction.INSTANCE);
this.functions.add(XxHashFunction.INSTANCE);
this.functions.add(WriteLogFunction.INSTANCE);
this.udf = new HashMap<>();
Expand Down Expand Up @@ -602,6 +616,81 @@ private ParseTimestampFunction() {
}
}

/** JSON_EACH_&lt;T&gt;(variant) returns a MAP&lt;VARCHAR, T&gt; holding all fields of
* a variant object whose values have the runtime type T. Fields with values
* of other types are not present in the result. Since JSON has no date or
* time types, the DATE, TIME, and TIMESTAMP functions also accept strings,
* parsed using the grammar of the corresponding SQL literal. A variant
* that does not hold an object produces an empty map. */
static class JsonEachFunction extends NonOptimizedFunction {
static final JsonEachFunction BIGINT = new JsonEachFunction("JSON_EACH_BIGINT", SqlTypeName.BIGINT);
static final JsonEachFunction STRING = new JsonEachFunction("JSON_EACH_STRING", SqlTypeName.VARCHAR);
static final JsonEachFunction BOOLEAN = new JsonEachFunction("JSON_EACH_BOOLEAN", SqlTypeName.BOOLEAN);
static final JsonEachFunction DATE = new JsonEachFunction("JSON_EACH_DATE", SqlTypeName.DATE);
static final JsonEachFunction TIME = new JsonEachFunction("JSON_EACH_TIME", SqlTypeName.TIME);
static final JsonEachFunction TIMESTAMP = new JsonEachFunction("JSON_EACH_TIMESTAMP", SqlTypeName.TIMESTAMP);

private JsonEachFunction(String name, SqlTypeName valueTypeName) {
super(name,
opBinding -> mapReturnType(opBinding, valueTypeName),
OperandTypes.VARIANT,
SqlFunctionCategory.USER_DEFINED_FUNCTION,
"json#json_each", FunctionDocumentation.NO_FILE);
}

/** MAP&lt;VARCHAR, valueTypeName&gt; with nullable values;
* the map itself is nullable iff the argument is. */
private static RelDataType mapReturnType(SqlOperatorBinding opBinding, SqlTypeName valueTypeName) {
RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
RelDataType valueType = typeFactory.createTypeWithNullability(
typeFactory.createSqlType(valueTypeName), true);
RelDataType mapType = typeFactory.createMapType(keyType, valueType);
return typeFactory.createTypeWithNullability(
mapType, opBinding.getOperandType(0).isNullable());
}
}

/** Functions returning the keys of a variant object as ARRAY&lt;VARCHAR&gt;.
* A variant that does not hold an object produces an empty array. */
static class JsonKeysFunction extends NonOptimizedFunction {
static final JsonKeysFunction OBJECT_KEYS =
new JsonKeysFunction("JSON_OBJECT_KEYS", "json#json_object_keys");
static final JsonKeysFunction KEYS =
new JsonKeysFunction("JSON_KEYS", "json#json_keys");

private JsonKeysFunction(String name, String documentation) {
super(name,
JsonKeysFunction::arrayReturnType,
OperandTypes.VARIANT,
SqlFunctionCategory.USER_DEFINED_FUNCTION,
documentation, FunctionDocumentation.NO_FILE);
}

/** ARRAY&lt;VARCHAR&gt; with non-null elements;
* the array itself is nullable iff the argument is. */
private static RelDataType arrayReturnType(SqlOperatorBinding opBinding) {
RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
RelDataType elementType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
RelDataType arrayType = typeFactory.createArrayType(elementType, -1);
return typeFactory.createTypeWithNullability(
arrayType, opBinding.getOperandType(0).isNullable());
}
}

static class VariantMergeFunction extends NonOptimizedFunction {
static final VariantMergeFunction INSTANCE = new VariantMergeFunction();

private VariantMergeFunction() {
super("VARIANT_MERGE",
ReturnTypes.VARIANT.andThen(SqlTypeTransforms.TO_NULLABLE),
OperandTypes.sequence("VARIANT_MERGE(<VARIANT>, <VARIANT>)",
OperandTypes.VARIANT, OperandTypes.VARIANT),
SqlFunctionCategory.USER_DEFINED_FUNCTION,
"json#variant_merge", FunctionDocumentation.NO_FILE);
}
}

/** RLIKE used as a function. RLIKE in SQL uses infix notation */
static class RlikeFunction extends NonOptimizedFunction {
static final RlikeFunction INSTANCE = new RlikeFunction();
Expand Down
84 changes: 84 additions & 0 deletions ...in/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/VariantFilterFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
package org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler;

import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.sql.SqlCallBinding;
import org.apache.calcite.sql.SqlFunctionCategory;
import org.apache.calcite.sql.SqlOperandCountRange;
import org.apache.calcite.sql.SqlOperator;
import org.apache.calcite.sql.type.FunctionSqlType;
import org.apache.calcite.sql.type.OperandTypes;
import org.apache.calcite.sql.type.SqlOperandCountRanges;
import org.apache.calcite.sql.type.SqlOperandTypeChecker;
import org.apache.calcite.sql.type.SqlReturnTypeInference;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.sql.type.SqlTypeUtil;

/** Calcite-level implementation of the VARIANT_FILTER and VARIANT_DEEP_FILTER functions.
* Both take (variant, (label, value) -&gt; predicate) and return a VARIANT with
* the items of the input variant for which the predicate is true. */
class VariantFilterFunction extends CustomFunctions.NonOptimizedFunction {
private VariantFilterFunction(String name, SqlTypeName labelTypeName, String documentation) {
super(name,
FILTER_INFERENCE,
makeChecker(name, labelTypeName),
SqlFunctionCategory.USER_DEFINED_FUNCTION,
documentation, FunctionDocumentation.NO_FILE);
}

/** Always nullable: a dropped non-map variant produces NULL */
static final SqlReturnTypeInference FILTER_INFERENCE = opBinding -> {
RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
return typeFactory.createTypeWithNullability(
typeFactory.createSqlType(SqlTypeName.VARIANT), true);
};

/** Checks (VARIANT, (labelTypeName, VARIANT) -&gt; BOOLEAN) operands */
static SqlOperandTypeChecker makeChecker(String name, SqlTypeName labelTypeName) {
String signature = name + "(<VARIANT>, <FUNCTION(" +
labelTypeName + ", VARIANT)-><BOOLEAN>>)";
return new SqlOperandTypeChecker() {
@Override
public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
if (!OperandTypes.VARIANT.checkSingleOperandType(
callBinding, callBinding.operand(0), 0, throwOnFailure))
return false;

RelDataTypeFactory typeFactory = callBinding.getTypeFactory();
// The label is NULL when the variant does not hold a map
RelDataType labelType = typeFactory.createTypeWithNullability(
typeFactory.createSqlType(labelTypeName), true);
RelDataType valueType = typeFactory.createSqlType(SqlTypeName.VARIANT);
GenericLambdaTypeChecker lambdaChecker =
new GenericLambdaTypeChecker(signature, labelType, valueType);
if (!lambdaChecker.checkSingleOperandType(
callBinding, callBinding.operand(1), 1, throwOnFailure))
return false;

RelDataType functionType = SqlTypeUtil.deriveType(callBinding, callBinding.operand(1));
if (!(functionType instanceof FunctionSqlType fType)
|| fType.getReturnType().getSqlTypeName() != SqlTypeName.BOOLEAN) {
if (throwOnFailure)
throw callBinding.newValidationSignatureError();
return false;
}
return true;
}

@Override
public SqlOperandCountRange getOperandCountRange() {
return SqlOperandCountRanges.of(2);
}

@Override
public String getAllowedSignatures(SqlOperator op, String opName) {
return signature;
}
};
}

static final VariantFilterFunction INSTANCE = new VariantFilterFunction(
"VARIANT_FILTER", SqlTypeName.VARIANT, "json#variant_filter");
static final VariantFilterFunction DEEP = new VariantFilterFunction(
"VARIANT_DEEP_FILTER", SqlTypeName.VARCHAR, "json#variant_deep_filter");
}
Loading
Loading