Skip to content
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

[core] Refactor the Field Aggregator factory to validate data types when creating function #4446

Merged
merged 2 commits into from
Nov 5, 2024
Merged

[core] Refactor the Field Aggregator factory to validate data types when creating function #4446

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2a7bc0a
Refactor the Field Aggregator factory to validate data types when cre…
zhuangchong Nov 4, 2024
98e746d
fix code style
zhuangchong Nov 4, 2024
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
6 changes: 3 additions & 3 deletions ...on-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
import org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
Expand Down Expand Up @@ -51,7 +52,6 @@
import static org.apache.paimon.CoreOptions.FIELDS_PREFIX;
import static org.apache.paimon.CoreOptions.FIELDS_SEPARATOR;
import static org.apache.paimon.CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE;
import static org.apache.paimon.mergetree.compact.aggregate.FieldAggregator.createFieldAggregator;
import static org.apache.paimon.utils.InternalRowUtils.createFieldGetters;

/**
Expand Down Expand Up @@ -495,7 +495,7 @@ private Map<Integer, Supplier<FieldAggregator>> createFieldAggregators(
fieldAggregators.put(
i,
() ->
createFieldAggregator(
FieldAggregatorFactory.create(
fieldType,
strAggFunc,
ignoreRetract,
Expand All @@ -506,7 +506,7 @@ private Map<Integer, Supplier<FieldAggregator>> createFieldAggregators(
fieldAggregators.put(
i,
() ->
createFieldAggregator(
FieldAggregatorFactory.create(
fieldType,
defaultAggFunc,
ignoreRetract,
Expand Down
3 changes: 2 additions & 1 deletion ...e/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.mergetree.compact.MergeFunction;
import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
import org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowKind;
Expand Down Expand Up @@ -142,7 +143,7 @@ public MergeFunction<KeyValue> create(@Nullable int[][] projection) {

boolean ignoreRetract = options.fieldAggIgnoreRetract(fieldName);
fieldAggregators[i] =
FieldAggregator.createFieldAggregator(
FieldAggregatorFactory.create(
fieldType,
strAggFunc,
ignoreRetract,
Expand Down
51 changes: 6 additions & 45 deletions paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,62 +18,23 @@

package org.apache.paimon.mergetree.compact.aggregate;

import org.apache.paimon.CoreOptions;
import org.apache.paimon.factories.FactoryUtil;
import org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
import org.apache.paimon.types.DataType;

import javax.annotation.Nullable;

import java.io.Serializable;

/** abstract class of aggregating a field of a row. */
public abstract class FieldAggregator implements Serializable {
protected DataType fieldType;

private static final long serialVersionUID = 1L;

public FieldAggregator(DataType dataType) {
this.fieldType = dataType;
}

public static FieldAggregator createFieldAggregator(
DataType fieldType,
@Nullable String strAgg,
boolean ignoreRetract,
boolean isPrimaryKey,
CoreOptions options,
String field) {
FieldAggregator fieldAggregator;
if (isPrimaryKey) {
strAgg = FieldPrimaryKeyAgg.NAME;
} else if (strAgg == null) {
strAgg = FieldLastNonNullValueAgg.NAME;
}

FieldAggregatorFactory fieldAggregatorFactory =
FactoryUtil.discoverFactory(
FieldAggregator.class.getClassLoader(),
FieldAggregatorFactory.class,
strAgg);
if (fieldAggregatorFactory == null) {
throw new RuntimeException(
String.format(
"Use unsupported aggregation: %s or spell aggregate function incorrectly!",
strAgg));
}
protected final DataType fieldType;
protected final String name;

fieldAggregator = fieldAggregatorFactory.create(fieldType, options, field);

if (ignoreRetract) {
fieldAggregator = new FieldIgnoreRetractAgg(fieldAggregator);
}

return fieldAggregator;
public FieldAggregator(String name, DataType dataType) {
this.name = name;
this.fieldType = dataType;
}

public abstract String name();

public abstract Object agg(Object accumulator, Object inputField);

public Object aggReversed(Object accumulator, Object inputField) {
Expand All @@ -89,6 +50,6 @@ public Object retract(Object accumulator, Object retractField) {
"Aggregate function '%s' does not support retraction,"
+ " If you allow this function to ignore retraction messages,"
+ " you can configure 'fields.${field_name}.ignore-retract'='true'.",
name()));
name));
}
}
31 changes: 5 additions & 26 deletions paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldBoolAndAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,43 +18,22 @@

package org.apache.paimon.mergetree.compact.aggregate;

import org.apache.paimon.types.DataType;
import org.apache.paimon.types.BooleanType;

/** bool_and aggregate a field of a row. */
public class FieldBoolAndAgg extends FieldAggregator {

public static final String NAME = "bool_and";

private static final long serialVersionUID = 1L;

public FieldBoolAndAgg(DataType dataType) {
super(dataType);
}

@Override
public String name() {
return NAME;
public FieldBoolAndAgg(String name, BooleanType dataType) {
super(name, dataType);
}

@Override
public Object agg(Object accumulator, Object inputField) {
Object boolAnd;

if (accumulator == null || inputField == null) {
boolAnd = (inputField == null) ? accumulator : inputField;
} else {
switch (fieldType.getTypeRoot()) {
case BOOLEAN:
boolAnd = (boolean) accumulator && (boolean) inputField;
break;
default:
String msg =
String.format(
"type %s not support in %s",
fieldType.getTypeRoot().toString(), this.getClass().getName());
throw new IllegalArgumentException(msg);
}
return accumulator == null ? inputField : accumulator;
}
return boolAnd;
return (boolean) accumulator && (boolean) inputField;
}
}
31 changes: 5 additions & 26 deletions paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldBoolOrAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,43 +18,22 @@

package org.apache.paimon.mergetree.compact.aggregate;

import org.apache.paimon.types.DataType;
import org.apache.paimon.types.BooleanType;

/** bool_or aggregate a field of a row. */
public class FieldBoolOrAgg extends FieldAggregator {

public static final String NAME = "bool_or";

private static final long serialVersionUID = 1L;

public FieldBoolOrAgg(DataType dataType) {
super(dataType);
}

@Override
public String name() {
return NAME;
public FieldBoolOrAgg(String name, BooleanType dataType) {
super(name, dataType);
}

@Override
public Object agg(Object accumulator, Object inputField) {
Object boolOr;

if (accumulator == null || inputField == null) {
boolOr = (inputField == null) ? accumulator : inputField;
} else {
switch (fieldType.getTypeRoot()) {
case BOOLEAN:
boolOr = (boolean) accumulator || (boolean) inputField;
break;
default:
String msg =
String.format(
"type %s not support in %s",
fieldType.getTypeRoot().toString(), this.getClass().getName());
throw new IllegalArgumentException(msg);
}
return accumulator == null ? inputField : accumulator;
}
return boolOr;
return (boolean) accumulator || (boolean) inputField;
}
}
11 changes: 2 additions & 9 deletions paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,16 +43,14 @@
/** Collect elements into an ARRAY. */
public class FieldCollectAgg extends FieldAggregator {

public static final String NAME = "collect";

private static final long serialVersionUID = 1L;

private final boolean distinct;
private final InternalArray.ElementGetter elementGetter;
@Nullable private final BiFunction<Object, Object, Boolean> equaliser;

public FieldCollectAgg(ArrayType dataType, boolean distinct) {
super(dataType);
public FieldCollectAgg(String name, ArrayType dataType, boolean distinct) {
super(name, dataType);
this.distinct = distinct;
this.elementGetter = InternalArray.createElementGetter(dataType.getElementType());

Expand Down Expand Up @@ -84,11 +82,6 @@ public FieldCollectAgg(ArrayType dataType, boolean distinct) {
}
}

@Override
public String name() {
return NAME;
}

@Override
public Object aggReversed(Object accumulator, Object inputField) {
// we don't need to actually do the reverse here for this agg
Expand Down
12 changes: 2 additions & 10 deletions ...rc/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,20 +23,12 @@
/** first non-null value aggregate a field of a row. */
public class FieldFirstNonNullValueAgg extends FieldAggregator {

public static final String NAME = "first_non_null_value";
public static final String LEGACY_NAME = "first_not_null_value";

private static final long serialVersionUID = 1L;

private boolean initialized;

public FieldFirstNonNullValueAgg(DataType dataType) {
super(dataType);
}

@Override
public String name() {
return NAME;
public FieldFirstNonNullValueAgg(String name, DataType dataType) {
super(name, dataType);
}

@Override
Expand Down
11 changes: 2 additions & 9 deletions ...-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstValueAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,19 +23,12 @@
/** first value aggregate a field of a row. */
public class FieldFirstValueAgg extends FieldAggregator {

public static final String NAME = "first_value";

private static final long serialVersionUID = 1L;

private boolean initialized;

public FieldFirstValueAgg(DataType dataType) {
super(dataType);
}

@Override
public String name() {
return NAME;
public FieldFirstValueAgg(String name, DataType dataType) {
super(name, dataType);
}

@Override
Expand Down
15 changes: 2 additions & 13 deletions ...n-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldHllSketchAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,25 +24,14 @@
/** HllSketch aggregate a field of a row. */
public class FieldHllSketchAgg extends FieldAggregator {

public static final String NAME = "hll_sketch";

private static final long serialVersionUID = 1L;

public FieldHllSketchAgg(VarBinaryType dataType) {
super(dataType);
}

@Override
public String name() {
return NAME;
public FieldHllSketchAgg(String name, VarBinaryType dataType) {
super(name, dataType);
}

@Override
public Object agg(Object accumulator, Object inputField) {
if (accumulator == null && inputField == null) {
return null;
}

if (accumulator == null || inputField == null) {
return accumulator == null ? inputField : accumulator;
}
Expand Down
11 changes: 3 additions & 8 deletions ...re/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,20 +21,15 @@
/** An aggregator which ignores retraction messages. */
public class FieldIgnoreRetractAgg extends FieldAggregator {

private final FieldAggregator aggregator;

private static final long serialVersionUID = 1L;

private final FieldAggregator aggregator;

public FieldIgnoreRetractAgg(FieldAggregator aggregator) {
super(aggregator.fieldType);
super(aggregator.name, aggregator.fieldType);
this.aggregator = aggregator;
}

@Override
public String name() {
return aggregator.name();
}

@Override
public Object agg(Object accumulator, Object inputField) {
return aggregator.agg(accumulator, inputField);
Expand Down
11 changes: 2 additions & 9 deletions ...src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastNonNullValueAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,17 +23,10 @@
/** last non-null value aggregate a field of a row. */
public class FieldLastNonNullValueAgg extends FieldAggregator {

public static final String NAME = "last_non_null_value";

private static final long serialVersionUID = 1L;

public FieldLastNonNullValueAgg(DataType dataType) {
super(dataType);
}

@Override
public String name() {
return NAME;
public FieldLastNonNullValueAgg(String name, DataType dataType) {
super(name, dataType);
}

@Override
Expand Down
11 changes: 2 additions & 9 deletions ...n-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,17 +23,10 @@
/** last value aggregate a field of a row. */
public class FieldLastValueAgg extends FieldAggregator {

public static final String NAME = "last_value";

private static final long serialVersionUID = 1L;

public FieldLastValueAgg(DataType dataType) {
super(dataType);
}

@Override
public String name() {
return NAME;
public FieldLastValueAgg(String name, DataType dataType) {
super(name, dataType);
}

@Override
Expand Down
Loading
Loading