Skip to content

Commit

Permalink
[INLONG-11260][SDK] Transform SQL supports "num_nonnulls" and "num_nu…
Browse files Browse the repository at this point in the history
…lls" function
  • Loading branch information
ZKpLo committed Oct 4, 2024
1 parent b709fc2 commit 1aeb867
Show file tree
Hide file tree
Showing 4 changed files with 240 additions and 0 deletions.
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
/*
* 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.inlong.sdk.transform.process.function;

import org.apache.inlong.sdk.transform.decode.SourceData;
import org.apache.inlong.sdk.transform.process.Context;
import org.apache.inlong.sdk.transform.process.operator.OperatorTools;
import org.apache.inlong.sdk.transform.process.parser.ValueParser;

import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.expression.Function;
import net.sf.jsqlparser.expression.NullValue;

import java.util.ArrayList;
import java.util.List;

/**
* NumNonNullsFunction -> num_nonnulls(expr1,expr2,...)
* description:
* - return the number of non-null arguments.
*/
@TransformFunction(names = {"num_nonnulls"})
public class NumNonNullsFunction implements ValueParser {

private final List<ValueParser> nodeList;

public NumNonNullsFunction(Function expr) {
List<Expression> expressions = expr.getParameters().getExpressions();
this.nodeList = new ArrayList<>();
if (expressions == null || expressions.isEmpty()) {
return;
}
for (Expression expression : expressions) {
if (expression instanceof NullValue) {
nodeList.add(null);
} else {
nodeList.add(OperatorTools.buildParser(expression));
}
}
}

@Override
public Object parse(SourceData sourceData, int rowIndex, Context context) {
int num = 0;
for (ValueParser valueParser : nodeList) {
if (valueParser == null) {
continue;
}
Object value = valueParser.parse(sourceData, rowIndex, context);
if (value != null) {
num++;
}
}
return num;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
/*
* 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.inlong.sdk.transform.process.function;

import org.apache.inlong.sdk.transform.decode.SourceData;
import org.apache.inlong.sdk.transform.process.Context;
import org.apache.inlong.sdk.transform.process.operator.OperatorTools;
import org.apache.inlong.sdk.transform.process.parser.ValueParser;

import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.expression.Function;

import java.util.ArrayList;
import java.util.List;

/**
* NumNonNullsFunction -> num_nulls(expr1,expr2,...)
* description:
* - return the number of null arguments.
*/
@TransformFunction(names = {"num_nulls"})
public class NumNullsFunction implements ValueParser {

private final List<ValueParser> nodeList;

public NumNullsFunction(Function expr) {
List<Expression> expressions = expr.getParameters().getExpressions();
this.nodeList = new ArrayList<>();
if (expressions == null || expressions.isEmpty()) {
return;
}
for (Expression expression : expressions) {
nodeList.add(OperatorTools.buildParser(expression));
}
}

@Override
public Object parse(SourceData sourceData, int rowIndex, Context context) {
int num = 0;
for (ValueParser valueParser : nodeList) {
if (valueParser == null) {
num++;
continue;
}
Object value = valueParser.parse(sourceData, rowIndex, context);
if (value == null) {
num++;
}
}
return num;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
/*
* 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.inlong.sdk.transform.process.function.arithmetic;

import org.apache.inlong.sdk.transform.decode.SourceDecoderFactory;
import org.apache.inlong.sdk.transform.encode.SinkEncoderFactory;
import org.apache.inlong.sdk.transform.pojo.TransformConfig;
import org.apache.inlong.sdk.transform.process.TransformProcessor;

import org.junit.Assert;
import org.junit.Test;

import java.util.HashMap;
import java.util.List;

public class TestNumNonNullsFunction extends AbstractFunctionArithmeticTestBase {

@Test
public void testNumNonNullsFunction() throws Exception {
String transformSql = null, data = null;
TransformConfig config = null;
TransformProcessor<String, String> processor = null;
List<String> output = null;

// case1: num_nonnulls(5, 3, null, null)
transformSql = "select num_nonnulls(numeric1,numeric2,null,numericx) from source";
config = new TransformConfig(transformSql);
processor = TransformProcessor
.create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
SinkEncoderFactory.createKvEncoder(kvSink));
data = "5|3|3|5";
output = processor.transform(data, new HashMap<>());
Assert.assertEquals(1, output.size());
Assert.assertEquals("result=2", output.get(0));
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
/*
* 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.inlong.sdk.transform.process.function.arithmetic;

import org.apache.inlong.sdk.transform.decode.SourceDecoderFactory;
import org.apache.inlong.sdk.transform.encode.SinkEncoderFactory;
import org.apache.inlong.sdk.transform.pojo.TransformConfig;
import org.apache.inlong.sdk.transform.process.TransformProcessor;

import org.junit.Assert;
import org.junit.Test;

import java.util.HashMap;
import java.util.List;

public class TestNumNullsFunction extends AbstractFunctionArithmeticTestBase {

@Test
public void testNumNullsFunction() throws Exception {
String transformSql = null, data = null;
TransformConfig config = null;
TransformProcessor<String, String> processor = null;
List<String> output = null;

// case1: num_nulls(5, null, null, null)
transformSql = "select num_nulls(numeric1,numericx1,null,numericx2) from source";
config = new TransformConfig(transformSql);
processor = TransformProcessor
.create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
SinkEncoderFactory.createKvEncoder(kvSink));
data = "5|3|3|5";
output = processor.transform(data, new HashMap<>());
Assert.assertEquals(1, output.size());
Assert.assertEquals("result=3", output.get(0));
}
}

0 comments on commit 1aeb867

Please sign in to comment.