You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2024/11/05 07:47:34 UTC
(flink) branch master updated: [FLINK-36615] Support LEAD/LAG functions in Table API (#25582)
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 96b79eb2373 [FLINK-36615] Support LEAD/LAG functions in Table API (#25582)
96b79eb2373 is described below
commit 96b79eb23738a52ec17918c47f14fe008431f18a
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Tue Nov 5 08:47:22 2024 +0100
[FLINK-36615] Support LEAD/LAG functions in Table API (#25582)
---
flink-python/pyflink/table/expressions.py | 22 ++++
.../org/apache/flink/table/api/Expressions.java | 120 +++++++++++++++++++++
.../org/apache/flink/table/api/OverWindow.java | 16 +--
.../table/api/OverWindowPartitionedOrdered.java | 9 +-
.../api/OverWindowPartitionedOrderedPreceding.java | 9 +-
.../expressions/resolver/ExpressionResolver.java | 2 +-
.../expressions/resolver/LocalOverWindow.java | 8 +-
.../resolver/rules/OverWindowResolverRule.java | 41 ++++---
.../functions/BuiltInFunctionDefinitions.java | 18 ++++
.../flink/table/functions/CallSyntaxUtils.java | 6 +-
.../flink/table/functions/SqlCallSyntax.java | 4 +-
.../strategies/LeadLagInputTypeStrategy.java | 109 +++++++++++++++++++
.../inference/strategies/OverTypeStrategy.java | 40 ++++---
.../strategies/SpecificInputTypeStrategies.java | 6 ++
.../strategies/SpecificTypeStrategies.java | 19 ++++
.../table/types/inference/TypeStrategiesTest.java | 20 +++-
.../strategies/LeadLagInputTypeStrategyTest.java | 55 ++++++++++
.../planner/expressions/SqlAggFunctionVisitor.java | 3 +
.../expressions/converter/OverConvertRule.java | 23 ++--
.../table/api/QueryOperationSqlExecutionTest.java | 1 +
.../api/QueryOperationSqlSerializationTest.java | 1 +
.../table/api/QueryOperationTestPrograms.java | 41 +++++++
22 files changed, 509 insertions(+), 64 deletions(-)
diff --git a/flink-python/pyflink/table/expressions.py b/flink-python/pyflink/table/expressions.py
index 25c16c81ba0..3b326cb4641 100644
--- a/flink-python/pyflink/table/expressions.py
+++ b/flink-python/pyflink/table/expressions.py
@@ -867,6 +867,28 @@ def json_array_agg(on_null: JsonOnNull, item_expr) -> Expression:
return _binary_op("jsonArrayAgg", on_null._to_j_json_on_null(), item_expr)
+def lag(expr, offset=1, default=None) -> Expression:
+ """
+ A window function that provides access to a row at a specified physical offset which comes
+ before the current row.
+ """
+ if default is None:
+ return _binary_op("lag", expr, offset)
+ else:
+ return _ternary_op("lag", expr, offset, default)
+
+
+def lead(expr, offset=1, default=None) -> Expression:
+ """
+ A window function that provides access to a row at a specified physical offset which comes
+ after the current row.
+ """
+ if default is None:
+ return _binary_op("lead", expr, offset)
+ else:
+ return _ternary_op("lead", expr, offset, default)
+
+
def call(f: Union[str, UserDefinedFunctionWrapper], *args) -> Expression:
"""
The first parameter `f` could be a str or a Python user-defined function.
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
index fbf4b930c60..022c137d2ba 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
@@ -947,6 +947,126 @@ public final class Expressions {
return apiCall(functionDefinition, itemExpr);
}
+ /**
+ * A window function that provides access to a row that comes directly after the current row.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * table.window(Over.orderBy($("ts")).partitionBy("organisation").as("w"))
+ * .select(
+ * $("organisation"),
+ * $("revenue"),
+ * lag($("revenue")).over($("w").as("next_revenue")
+ * )
+ * }</pre>
+ */
+ public static ApiExpression lead(Object value) {
+ return apiCall(BuiltInFunctionDefinitions.LEAD, value);
+ }
+
+ /**
+ * A window function that provides access to a row at a specified physical offset which comes
+ * after the current row.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * table.window(Over.orderBy($("ts")).partitionBy("organisation").as("w"))
+ * .select(
+ * $("organisation"),
+ * $("revenue"),
+ * lag($("revenue"), 1).over($("w").as("next_revenue")
+ * )
+ * }</pre>
+ */
+ public static ApiExpression lead(Object value, Object offset) {
+ return apiCall(BuiltInFunctionDefinitions.LEAD, value, offset);
+ }
+
+ /**
+ * A window function that provides access to a row at a specified physical offset which comes
+ * after the current row.
+ *
+ * <p>The value to return when offset is beyond the scope of the partition. If a default value
+ * is not specified, NULL is returned. {@code default} must be type-compatible with {@code
+ * value}.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * table.window(Over.orderBy($("ts")).partitionBy("organisation").as("w"))
+ * .select(
+ * $("organisation"),
+ * $("revenue"),
+ * lag($("revenue"), 1, lit(0)).over($("w").as("next_revenue")
+ * )
+ * }</pre>
+ */
+ public static ApiExpression lead(Object value, Object offset, Object defaultValue) {
+ return apiCall(BuiltInFunctionDefinitions.LEAD, value, offset, defaultValue);
+ }
+
+ /**
+ * A window function that provides access to a row that comes directly before the current row.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * table.window(Over.orderBy($("ts")).partitionBy("organisation").as("w"))
+ * .select(
+ * $("organisation"),
+ * $("revenue"),
+ * lag($("revenue")).over($("w").as("prev_revenue")
+ * )
+ * }</pre>
+ */
+ public static ApiExpression lag(Object value) {
+ return apiCall(BuiltInFunctionDefinitions.LAG, value);
+ }
+
+ /**
+ * A window function that provides access to a row at a specified physical offset which comes
+ * before the current row.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * table.window(Over.orderBy($("ts")).partitionBy("organisation").as("w"))
+ * .select(
+ * $("organisation"),
+ * $("revenue"),
+ * lag($("revenue"), 1).over($("w").as("prev_revenue")
+ * )
+ * }</pre>
+ */
+ public static ApiExpression lag(Object value, Object offset) {
+ return apiCall(BuiltInFunctionDefinitions.LAG, value, offset);
+ }
+
+ /**
+ * A window function that provides access to a row at a specified physical offset which comes
+ * before the current row.
+ *
+ * <p>The value to return when offset is beyond the scope of the partition. If a default value
+ * is not specified, NULL is returned. {@code default} must be type-compatible with {@code
+ * value}.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * org.window(Over.orderBy($("ts")).partitionBy("organisation").as("w"))
+ * .select(
+ * $("organisation"),
+ * $("revenue"),
+ * lag($("revenue"), 1, lit(0)).over($("w").as("prev_revenue")
+ * )
+ * }</pre>
+ */
+ public static ApiExpression lag(Object value, Object offset, Object defaultValue) {
+ return apiCall(BuiltInFunctionDefinitions.LAG, value, offset, defaultValue);
+ }
+
/**
* A call to a function that will be looked up in a catalog. There are two kinds of functions:
*
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindow.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindow.java
index def4db27f23..ab57d254c08 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindow.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindow.java
@@ -21,6 +21,8 @@ package org.apache.flink.table.api;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.expressions.Expression;
+import javax.annotation.Nullable;
+
import java.util.List;
import java.util.Optional;
@@ -36,15 +38,15 @@ public final class OverWindow {
private final Expression alias;
private final List<Expression> partitioning;
private final Expression order;
- private final Expression preceding;
- private final Optional<Expression> following;
+ private final @Nullable Expression preceding;
+ private final @Nullable Expression following;
OverWindow(
Expression alias,
List<Expression> partitionBy,
Expression orderBy,
- Expression preceding,
- Optional<Expression> following) {
+ @Nullable Expression preceding,
+ @Nullable Expression following) {
this.alias = alias;
this.partitioning = partitionBy;
this.order = orderBy;
@@ -64,11 +66,11 @@ public final class OverWindow {
return order;
}
- public Expression getPreceding() {
- return preceding;
+ public Optional<Expression> getPreceding() {
+ return Optional.ofNullable(preceding);
}
public Optional<Expression> getFollowing() {
- return following;
+ return Optional.ofNullable(following);
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java
index c4ed5657cd5..622ee2922ec 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java
@@ -22,10 +22,8 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.expressions.Expression;
import java.util.List;
-import java.util.Optional;
import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef;
-import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral;
/** Partially defined over window with (optional) partitioning and order. */
@PublicEvolving
@@ -66,11 +64,6 @@ public final class OverWindowPartitionedOrdered {
* @return the fully defined over window
*/
public OverWindow as(Expression alias) {
- return new OverWindow(
- alias,
- partitionBy,
- orderBy,
- valueLiteral(OverWindowRange.UNBOUNDED_RANGE),
- Optional.empty());
+ return new OverWindow(alias, partitionBy, orderBy, null, null);
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrderedPreceding.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrderedPreceding.java
index 34a3b5e9151..b4aeabcf505 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrderedPreceding.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrderedPreceding.java
@@ -21,8 +21,9 @@ package org.apache.flink.table.api;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.expressions.Expression;
+import javax.annotation.Nullable;
+
import java.util.List;
-import java.util.Optional;
import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef;
@@ -32,8 +33,8 @@ public final class OverWindowPartitionedOrderedPreceding {
private final List<Expression> partitionBy;
private final Expression orderBy;
- private final Expression preceding;
- private Optional<Expression> optionalFollowing = Optional.empty();
+ private final @Nullable Expression preceding;
+ private @Nullable Expression optionalFollowing = null;
OverWindowPartitionedOrderedPreceding(
List<Expression> partitionBy, Expression orderBy, Expression preceding) {
@@ -69,7 +70,7 @@ public final class OverWindowPartitionedOrderedPreceding {
* @return an over window with defined following
*/
public OverWindowPartitionedOrderedPreceding following(Expression following) {
- optionalFollowing = Optional.of(following);
+ optionalFollowing = following;
return this;
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/ExpressionResolver.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/ExpressionResolver.java
index d385a4ec74e..210e1dfd85d 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/ExpressionResolver.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/ExpressionResolver.java
@@ -367,7 +367,7 @@ public class ExpressionResolver {
overWindow.getAlias(),
prepareExpressions(overWindow.getPartitioning()),
resolveFieldsInSingleExpression(overWindow.getOrder()),
- resolveFieldsInSingleExpression(overWindow.getPreceding()),
+ overWindow.getPreceding().map(this::resolveFieldsInSingleExpression).orElse(null),
overWindow.getFollowing().map(this::resolveFieldsInSingleExpression).orElse(null));
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/LocalOverWindow.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/LocalOverWindow.java
index 01258b02d54..e2d88ac3464 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/LocalOverWindow.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/LocalOverWindow.java
@@ -36,7 +36,7 @@ public final class LocalOverWindow {
private Expression orderBy;
- private Expression preceding;
+ private @Nullable Expression preceding;
private @Nullable Expression following;
@@ -44,7 +44,7 @@ public final class LocalOverWindow {
Expression alias,
List<Expression> partitionBy,
Expression orderBy,
- Expression preceding,
+ @Nullable Expression preceding,
@Nullable Expression following) {
this.alias = alias;
this.partitionBy = partitionBy;
@@ -65,8 +65,8 @@ public final class LocalOverWindow {
return orderBy;
}
- public Expression getPreceding() {
- return preceding;
+ public Optional<Expression> getPreceding() {
+ return Optional.ofNullable(preceding);
}
public Optional<Expression> getFollowing() {
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java
index b6cf186ca68..ea7fe0dfe66 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.expressions.resolver.rules;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.OverWindowRange;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.expressions.Expression;
@@ -33,7 +34,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
-import static java.util.Arrays.asList;
import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedCall;
import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.BIGINT;
@@ -76,14 +76,29 @@ final class OverWindowResolverRule implements ResolverRule {
new ValidationException(
"Could not resolve over call."));
- Expression following = calculateOverWindowFollowing(referenceWindow);
- List<Expression> newArgs =
- new ArrayList<>(
- asList(
- children.get(0),
- referenceWindow.getOrderBy(),
- referenceWindow.getPreceding(),
- following));
+ UnresolvedCallExpression agg = (UnresolvedCallExpression) children.get(0);
+
+ final List<Expression> newArgs = new ArrayList<>();
+ newArgs.add(agg);
+ newArgs.add(referenceWindow.getOrderBy());
+ if (agg.getFunctionDefinition() == BuiltInFunctionDefinitions.LAG
+ || agg.getFunctionDefinition() == BuiltInFunctionDefinitions.LEAD) {
+ if (referenceWindow.getPreceding().isPresent()
+ || referenceWindow.getFollowing().isPresent()) {
+ throw new ValidationException(
+ "LEAD/LAG functions do not support "
+ + "providing RANGE/ROW bounds.");
+ }
+ newArgs.add(valueLiteral(null, DataTypes.NULL()));
+ newArgs.add(valueLiteral(null, DataTypes.NULL()));
+ } else {
+ Expression preceding =
+ referenceWindow
+ .getPreceding()
+ .orElse(valueLiteral(OverWindowRange.UNBOUNDED_RANGE));
+ newArgs.add(preceding);
+ newArgs.add(calculateOverWindowFollowing(referenceWindow, preceding));
+ }
newArgs.addAll(referenceWindow.getPartitionBy());
@@ -96,15 +111,13 @@ final class OverWindowResolverRule implements ResolverRule {
}
}
- private Expression calculateOverWindowFollowing(LocalOverWindow referenceWindow) {
+ private Expression calculateOverWindowFollowing(
+ LocalOverWindow referenceWindow, Expression preceding) {
return referenceWindow
.getFollowing()
.orElseGet(
() -> {
- WindowKind kind =
- referenceWindow
- .getPreceding()
- .accept(OVER_WINDOW_KIND_EXTRACTOR);
+ WindowKind kind = preceding.accept(OVER_WINDOW_KIND_EXTRACTOR);
if (kind == WindowKind.ROW) {
return valueLiteral(OverWindowRange.CURRENT_ROW);
} else {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 13dfdf8054b..5e62819495b 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -782,6 +782,24 @@ public final class BuiltInFunctionDefinitions {
.outputTypeStrategy(TypeStrategies.aggArg0(t -> t, true))
.build();
+ public static final BuiltInFunctionDefinition LEAD =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("lead")
+ .kind(AGGREGATE)
+ .inputTypeStrategy(SpecificInputTypeStrategies.LEAD_LAG)
+ .outputTypeStrategy(SpecificTypeStrategies.LEAD_LAG)
+ .runtimeDeferred()
+ .build();
+
+ public static final BuiltInFunctionDefinition LAG =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("lag")
+ .kind(AGGREGATE)
+ .inputTypeStrategy(SpecificInputTypeStrategies.LEAD_LAG)
+ .outputTypeStrategy(SpecificTypeStrategies.LEAD_LAG)
+ .runtimeDeferred()
+ .build();
+
public static final BuiltInFunctionDefinition LISTAGG =
BuiltInFunctionDefinition.newBuilder()
.name("listAgg")
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/CallSyntaxUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/CallSyntaxUtils.java
index 82c052d12a9..3ab541a825d 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/CallSyntaxUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/CallSyntaxUtils.java
@@ -53,8 +53,12 @@ class CallSyntaxUtils {
static String overRangeToSerializableString(
ResolvedExpression preceding, ResolvedExpression following) {
+ if (((ValueLiteralExpression) preceding).isNull()
+ || ((ValueLiteralExpression) following).isNull()) {
+ return "";
+ }
return String.format(
- "%s BETWEEN %s AND %s",
+ " %s BETWEEN %s AND %s",
isRowsRange(preceding) ? "ROWS" : "RANGE",
toStringPrecedingOrFollowing(preceding, true),
toStringPrecedingOrFollowing(following, false));
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java
index 96e47d6e8ad..f961391a07a 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java
@@ -325,10 +325,10 @@ public interface SqlCallSyntax {
CallSyntaxUtils.overRangeToSerializableString(
operands.get(2), operands.get(3));
if (operands.size() == 4) {
- return String.format("%s OVER(ORDER BY %s %s)", projection, order, rangeBounds);
+ return String.format("%s OVER(ORDER BY %s%s)", projection, order, rangeBounds);
} else {
return String.format(
- "%s OVER(PARTITION BY %s ORDER BY %s %s)",
+ "%s OVER(PARTITION BY %s ORDER BY %s%s)",
projection,
CallSyntaxUtils.asSerializableOperand(operands.get(4)),
order,
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LeadLagInputTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LeadLagInputTypeStrategy.java
new file mode 100644
index 00000000000..92ae27ab17d
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LeadLagInputTypeStrategy.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.ArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.ConstantArgumentCount;
+import org.apache.flink.table.types.inference.InputTypeStrategies;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Type strategy for {@link BuiltInFunctionDefinitions#LAG} and { @link
+ * BuiltInFunctionDefinitions#LEAD}.
+ *
+ * <p>The second argument needs to be NUMERIC if provided. Third argument must have a common type
+ * with the first.
+ */
+@Internal
+public final class LeadLagInputTypeStrategy implements InputTypeStrategy {
+
+ private static final ArgumentTypeStrategy NUMERIC_ARGUMENT =
+ InputTypeStrategies.logical(LogicalTypeFamily.NUMERIC);
+
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return ConstantArgumentCount.between(1, 3);
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ CallContext callContext, boolean throwOnFailure) {
+ List<DataType> argumentDataTypes = callContext.getArgumentDataTypes();
+
+ if (argumentDataTypes.size() == 1) {
+ return Optional.of(argumentDataTypes);
+ }
+
+ Optional<DataType> offsetArg =
+ NUMERIC_ARGUMENT.inferArgumentType(callContext, 1, throwOnFailure);
+
+ if (!offsetArg.isPresent()) {
+ return Optional.empty();
+ }
+
+ final DataType arg0 = argumentDataTypes.get(0);
+ if (argumentDataTypes.size() == 2) {
+ return Optional.of(Arrays.asList(arg0, offsetArg.get()));
+ } else {
+ LogicalType defaultType = argumentDataTypes.get(2).getLogicalType();
+ Optional<LogicalType> commonType =
+ LogicalTypeMerging.findCommonType(
+ Arrays.asList(arg0.getLogicalType(), defaultType));
+
+ if (!commonType.isPresent()) {
+ return callContext.fail(
+ throwOnFailure,
+ "The default value must have a common "
+ + "type with the given expression. ARG0: %s, default: %s",
+ arg0,
+ defaultType);
+ }
+
+ DataType commonDataType = commonType.map(TypeConversions::fromLogicalToDataType).get();
+ return Optional.of(Arrays.asList(commonDataType, offsetArg.get(), commonDataType));
+ }
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
+ return Arrays.asList(
+ Signature.of(Argument.ofGroup("ANY")),
+ Signature.of(Argument.ofGroup("ANY"), Argument.ofGroup("NUMERIC")),
+ Signature.of(
+ Argument.ofGroup("COMMON"),
+ Argument.ofGroup("NUMERIC"),
+ Argument.ofGroup("COMMON")));
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategy.java
index 3fd35498759..76109205263 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategy.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategy.java
@@ -28,11 +28,12 @@ import org.apache.flink.table.types.inference.CallContext;
import org.apache.flink.table.types.inference.ConstantArgumentCount;
import org.apache.flink.table.types.inference.InputTypeStrategy;
import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
import java.math.BigDecimal;
-import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Optional;
@@ -70,6 +71,27 @@ public class OverTypeStrategy implements InputTypeStrategy {
final DataType preceding = argumentDataTypes.get(2);
final DataType following = argumentDataTypes.get(3);
+ if (preceding.getLogicalType().is(LogicalTypeRoot.NULL)
+ || following.getLogicalType().is(LogicalTypeRoot.NULL)) {
+ if (!preceding.getLogicalType().is(LogicalTypeRoot.NULL)
+ || !following.getLogicalType().is(LogicalTypeRoot.NULL)) {
+ return callContext.fail(
+ throwOnFailure, "Both preceding and following must be provided" + ".");
+ }
+
+ return Optional.of(argumentDataTypes);
+ }
+
+ return validatePrecedingFollowingPresent(
+ callContext, throwOnFailure, preceding, following, argumentDataTypes);
+ }
+
+ private Optional<List<DataType>> validatePrecedingFollowingPresent(
+ CallContext callContext,
+ boolean throwOnFailure,
+ DataType preceding,
+ DataType following,
+ List<DataType> argumentDataTypes) {
final LogicalTypeRoot precedingTypeRoot = preceding.getLogicalType().getTypeRoot();
final Optional<String> precedingErr =
validateFollowingPreceding(
@@ -180,20 +202,6 @@ public class OverTypeStrategy implements InputTypeStrategy {
@Override
public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
- return Arrays.asList(
- Signature.of(
- Signature.Argument.ofGroup("ANY"),
- Signature.Argument.ofGroup("TIME ATTRIBUTE"),
- Signature.Argument.ofGroup(
- "INTERVAL LITERAL | CURRENT_RANGE | UNBOUNDED_RANGE"),
- Signature.Argument.ofGroup(
- "INTERVAL LITERAL | CURRENT_RANGE | UNBOUNDED_RANGE"),
- Signature.Argument.ofGroupVarying("ANY")),
- Signature.of(
- Signature.Argument.ofGroup("ANY"),
- Signature.Argument.ofGroup("TIME ATTRIBUTE"),
- Signature.Argument.ofGroup("BIGINT | CURRENT_ROW | UNBOUNDED_ROW"),
- Signature.Argument.ofGroup("BIGINT | CURRENT_ROW | UNBOUNDED_ROW"),
- Signature.Argument.ofGroupVarying("ANY")));
+ return Collections.singletonList(Signature.of(Argument.ofGroupVarying("INTERNAL")));
}
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
index 1c20f91c6ee..55e0760499a 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
@@ -167,6 +167,12 @@ public final class SpecificInputTypeStrategies {
/** Type strategy specific for {@link BuiltInFunctionDefinitions#IN}. */
public static final InputTypeStrategy IN = new SubQueryInputTypeStrategy();
+ /**
+ * Type strategy for {@link BuiltInFunctionDefinitions#LAG} and { @link
+ * BuiltInFunctionDefinitions#LEAD}.
+ */
+ public static final InputTypeStrategy LEAD_LAG = new LeadLagInputTypeStrategy();
+
private SpecificInputTypeStrategies() {
// no instantiation
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
index ed0fc8063ee..df8ac3eeb0f 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
@@ -22,11 +22,13 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.types.CollectionDataType;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.KeyValueDataType;
import org.apache.flink.table.types.inference.TypeStrategies;
import org.apache.flink.table.types.inference.TypeStrategy;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import java.util.List;
import java.util.Optional;
/**
@@ -169,6 +171,23 @@ public final class SpecificTypeStrategies {
((CollectionDataType) callContext.getArgumentDataTypes().get(1))
.getElementDataType()));
+ /**
+ * Strategy for {@link org.apache.flink.table.functions.BuiltInFunctionDefinitions#LAG} and
+ * {@link org.apache.flink.table.functions.BuiltInFunctionDefinitions#LEAD}. Returns a nullable
+ * type of arg0, unless the default value is not null. In that case the result will be not null.
+ */
+ public static final TypeStrategy LEAD_LAG =
+ callContext -> {
+ final List<DataType> argumentDataTypes = callContext.getArgumentDataTypes();
+ final DataType arg0 = argumentDataTypes.get(0);
+ if (argumentDataTypes.size() == 3
+ && !argumentDataTypes.get(2).getLogicalType().isNullable()) {
+ return Optional.of(arg0.notNull());
+ } else {
+ return Optional.of(arg0.nullable());
+ }
+ };
+
private SpecificTypeStrategies() {
// no instantiation
}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTest.java
index 797d330baff..1dddc69720f 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.types.inference;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies;
import org.apache.flink.table.types.logical.LogicalTypeFamily;
import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
@@ -178,6 +179,23 @@ class TypeStrategiesTest extends TypeStrategiesTestBase {
.expectDataType(DataTypes.DOUBLE()),
TypeStrategiesTestBase.TestSpec.forStrategy(PERCENTILE)
.inputTypes(DataTypes.INT(), DataTypes.ARRAY(DataTypes.DECIMAL(5, 2)))
- .expectDataType(DataTypes.ARRAY(DataTypes.DOUBLE())));
+ .expectDataType(DataTypes.ARRAY(DataTypes.DOUBLE())),
+
+ // LeadLagStrategy
+ TypeStrategiesTestBase.TestSpec.forStrategy(
+ "Expression not null", SpecificTypeStrategies.LEAD_LAG)
+ .inputTypes(DataTypes.INT().notNull(), DataTypes.BIGINT())
+ .expectDataType(DataTypes.INT()),
+ TypeStrategiesTestBase.TestSpec.forStrategy(
+ "Default value not null", SpecificTypeStrategies.LEAD_LAG)
+ .inputTypes(
+ DataTypes.STRING(),
+ DataTypes.BIGINT(),
+ DataTypes.STRING().notNull())
+ .expectDataType(DataTypes.STRING().notNull()),
+ TypeStrategiesTestBase.TestSpec.forStrategy(
+ "Default value nullable", SpecificTypeStrategies.LEAD_LAG)
+ .inputTypes(DataTypes.STRING(), DataTypes.BIGINT(), DataTypes.STRING())
+ .expectDataType(DataTypes.STRING()));
}
}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LeadLagInputTypeStrategyTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LeadLagInputTypeStrategyTest.java
new file mode 100644
index 00000000000..8e3b8de659b
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LeadLagInputTypeStrategyTest.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.inference.InputTypeStrategiesTestBase;
+
+import java.util.stream.Stream;
+
+/** Tests for {@link LeadLagInputTypeStrategy}. */
+class LeadLagInputTypeStrategyTest extends InputTypeStrategiesTestBase {
+
+ @Override
+ protected Stream<TestSpec> testData() {
+ return Stream.of(
+ TestSpec.forStrategy(SpecificInputTypeStrategies.LEAD_LAG)
+ .calledWithArgumentTypes(
+ DataTypes.BIGINT(), DataTypes.SMALLINT(), DataTypes.INT())
+ .calledWithLiteralAt(1)
+ .expectSignature(
+ "f(<ANY>)\n"
+ + "f(<ANY>, <NUMERIC>)\n"
+ + "f(<COMMON>, <NUMERIC>, <COMMON>)")
+ .expectArgumentTypes(
+ DataTypes.BIGINT(), DataTypes.SMALLINT(), DataTypes.BIGINT()),
+ TestSpec.forStrategy(SpecificInputTypeStrategies.LEAD_LAG)
+ .calledWithArgumentTypes(
+ DataTypes.BIGINT(), DataTypes.SMALLINT(), DataTypes.STRING())
+ .expectErrorMessage(
+ "The default value must have a"
+ + " common type with the given expression. ARG0: BIGINT, "
+ + "default: STRING"),
+ TestSpec.forStrategy(SpecificInputTypeStrategies.LEAD_LAG)
+ .calledWithArgumentTypes(DataTypes.BIGINT(), DataTypes.STRING())
+ .expectErrorMessage(
+ "Unsupported argument type. "
+ + "Expected type of family 'NUMERIC' but actual type was 'STRING'."));
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/SqlAggFunctionVisitor.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/SqlAggFunctionVisitor.java
index 24cde572c7e..ca3b67612a0 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/SqlAggFunctionVisitor.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/SqlAggFunctionVisitor.java
@@ -70,6 +70,9 @@ public class SqlAggFunctionVisitor extends ExpressionDefaultVisitor<SqlAggFuncti
BuiltInFunctionDefinitions.FIRST_VALUE, FlinkSqlOperatorTable.FIRST_VALUE);
AGG_DEF_SQL_OPERATOR_MAPPING.put(
BuiltInFunctionDefinitions.LAST_VALUE, FlinkSqlOperatorTable.LAST_VALUE);
+ AGG_DEF_SQL_OPERATOR_MAPPING.put(BuiltInFunctionDefinitions.LAG, FlinkSqlOperatorTable.LAG);
+ AGG_DEF_SQL_OPERATOR_MAPPING.put(
+ BuiltInFunctionDefinitions.LEAD, FlinkSqlOperatorTable.LEAD);
AGG_DEF_SQL_OPERATOR_MAPPING.put(
BuiltInFunctionDefinitions.LISTAGG, FlinkSqlOperatorTable.LISTAGG);
AGG_DEF_SQL_OPERATOR_MAPPING.put(
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java
index 46dbe94310e..2889d0ac2a2 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java
@@ -139,6 +139,9 @@ public class OverConvertRule implements CallExpressionConvertRule {
}
private static boolean isRows(ValueLiteralExpression preceding) {
+ if (preceding.isNull()) {
+ return true;
+ }
return preceding
.getValueAs(OverWindowRange.class)
.map(r -> r == OverWindowRange.CURRENT_ROW || r == OverWindowRange.UNBOUNDED_ROW)
@@ -193,12 +196,20 @@ public class OverConvertRule implements CallExpressionConvertRule {
private RexWindowBound createBound(ConvertContext context, Expression bound, SqlKind sqlKind) {
if (bound instanceof ValueLiteralExpression) {
final ValueLiteralExpression literal = (ValueLiteralExpression) bound;
- return literal.getValueAs(OverWindowRange.class)
- .map(r -> createSymbolBound(r, sqlKind))
- .orElseGet(
- () ->
- createLiteralBound(
- context, (ValueLiteralExpression) bound, sqlKind));
+ if (literal.isNull()) {
+ if (sqlKind == SqlKind.PRECEDING) {
+ return RexWindowBounds.UNBOUNDED_PRECEDING;
+ } else {
+ return RexWindowBounds.CURRENT_ROW;
+ }
+ } else {
+ return literal.getValueAs(OverWindowRange.class)
+ .map(r -> createSymbolBound(r, sqlKind))
+ .orElseGet(
+ () ->
+ createLiteralBound(
+ context, (ValueLiteralExpression) bound, sqlKind));
+ }
} else {
throw new TableException("Unexpected expression: " + bound);
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlExecutionTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlExecutionTest.java
index 0fac184fb42..2b658d2706c 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlExecutionTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlExecutionTest.java
@@ -75,6 +75,7 @@ public class QueryOperationSqlExecutionTest implements TableTestProgramRunner {
QueryOperationTestPrograms.OVER_WINDOW_RANGE,
QueryOperationTestPrograms.OVER_WINDOW_ROWS,
QueryOperationTestPrograms.OVER_WINDOW_ROWS_UNBOUNDED_NO_PARTITION,
+ QueryOperationTestPrograms.OVER_WINDOW_LAG,
QueryOperationTestPrograms.ACCESSING_NESTED_COLUMN);
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
index a006b1a46c4..f0eb882e6a1 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
@@ -64,6 +64,7 @@ public class QueryOperationSqlSerializationTest implements TableTestProgramRunne
QueryOperationTestPrograms.OVER_WINDOW_RANGE,
QueryOperationTestPrograms.OVER_WINDOW_ROWS,
QueryOperationTestPrograms.OVER_WINDOW_ROWS_UNBOUNDED_NO_PARTITION,
+ QueryOperationTestPrograms.OVER_WINDOW_LAG,
QueryOperationTestPrograms.ACCESSING_NESTED_COLUMN);
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
index a53d3d3eda2..d8e4ce8accd 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
@@ -31,11 +31,13 @@ import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
+import java.util.Collections;
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.UNBOUNDED_ROW;
import static org.apache.flink.table.api.Expressions.call;
import static org.apache.flink.table.api.Expressions.ifThenElse;
+import static org.apache.flink.table.api.Expressions.lag;
import static org.apache.flink.table.api.Expressions.lit;
import static org.apache.flink.table.api.Expressions.nullOf;
import static org.apache.flink.table.api.Expressions.row;
@@ -904,6 +906,45 @@ public class QueryOperationTestPrograms {
"sink")
.build();
+ static final TableTestProgram OVER_WINDOW_LAG =
+ TableTestProgram.of("over-window-lag", "validates over window with lag function")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "ts STRING",
+ "b MAP<DOUBLE, DOUBLE>",
+ "`r_time` AS TO_TIMESTAMP(`ts`)",
+ "WATERMARK for `r_time` AS `r_time`")
+ .producedBeforeRestore(
+ Row.of(
+ "2020-04-15 08:00:05",
+ Collections.singletonMap(42.0, 42.0)))
+ .producedAfterRestore(
+ Row.of(
+ "2020-04-15 08:00:06",
+ Collections.singletonMap(42.1, 42.1)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("ts STRING", "b MAP<DOUBLE, DOUBLE>")
+ .consumedBeforeRestore(Row.of("2020-04-15 08:00:05", null))
+ .consumedAfterRestore(
+ Row.of(
+ "2020-04-15 08:00:06",
+ Collections.singletonMap(42.0, 42.0)))
+ .build())
+ .runTableApi(
+ env ->
+ env.from("t")
+ .window(Over.orderBy($("r_time")).as("bLag"))
+ .select($("ts"), lag($("b"), 1).over($("bLag"))),
+ "sink_t")
+ .runSql(
+ "SELECT `$$T_PROJECT`.`ts`, (LAG(`$$T_PROJECT`.`b`, 1) OVER(ORDER BY `$$T_PROJECT`.`r_time`)) AS `_c1` FROM (\n"
+ + " SELECT `$$T_SOURCE`.`ts`, `$$T_SOURCE`.`b`, `$$T_SOURCE`.`r_time` FROM `default_catalog`.`default_database`.`t` $$T_SOURCE\n"
+ + ") $$T_PROJECT")
+ .build();
+
static final TableTestProgram ACCESSING_NESTED_COLUMN =
TableTestProgram.of(
"project-nested-columnd",