You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by xi...@apache.org on 2018/01/10 19:17:49 UTC
samza git commit: SAMZA-1535: Support for UDFs in where clauses
Repository: samza
Updated Branches:
refs/heads/master a6540b4e3 -> 9893069cb
SAMZA-1535: Support for UDFs in where clauses
The existing version of the udf implementation doesn't seem to support udfs in the where clauses because the Type of the object returned is "ANY" and when you do a
`select * from kafka.topic where regexMatch('.*foo', Name)` it fails in the query validation, because calcite doesn't know the type of regexMatch.
To solve the problem, We made the scalarUdf generic with a strongly typed return type.
https://issues.apache.org/jira/browse/SAMZA-1535
This PR can be merged into trunk not the 0.14.
Author: Srinivasulu Punuru <sp...@linkedin.com>
Reviewers: Xinyu Liu <xi...@gmail.com>
Closes #386 from srinipunuru/udf-where.1
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/9893069c
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/9893069c
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/9893069c
Branch: refs/heads/master
Commit: 9893069cb5186d58e3cf5518ce8e5a7251d8b131
Parents: a6540b4
Author: Srinivasulu Punuru <sp...@linkedin.com>
Authored: Wed Jan 10 11:17:39 2018 -0800
Committer: xiliu <xi...@xiliu-ld1.linkedin.biz>
Committed: Wed Jan 10 11:17:39 2018 -0800
----------------------------------------------------------------------
.../org/apache/samza/sql/udfs/ScalarUdf.java | 4 +-
.../apache/samza/sql/avro/AvroRelConverter.java | 2 +-
.../samza/sql/data/SamzaSqlRelMessage.java | 2 +
.../org/apache/samza/sql/fn/FlattenUdf.java | 2 +-
.../org/apache/samza/sql/fn/RegexMatchUdf.java | 39 +++++++++++++++++++
.../samza/sql/impl/ConfigBasedUdfResolver.java | 12 +++---
.../sql/planner/SamzaSqlScalarFunctionImpl.java | 7 +++-
.../samza/sql/avro/TestAvroRelConversion.java | 14 +++++++
.../samza/sql/e2e/TestSamzaSqlEndToEnd.java | 16 ++++++++
.../samza/sql/testutil/MyTestArrayUdf.java | 5 ++-
.../apache/samza/sql/testutil/MyTestUdf.java | 2 +-
.../samza/sql/testutil/SamzaSqlTestConfig.java | 6 ++-
.../org/apache/samza/tools/SamzaSqlConsole.java | 12 +++---
.../apache/samza/tools/udf/RegexMatchUdf.java | 40 --------------------
14 files changed, 100 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-api/src/main/java/org/apache/samza/sql/udfs/ScalarUdf.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/sql/udfs/ScalarUdf.java b/samza-api/src/main/java/org/apache/samza/sql/udfs/ScalarUdf.java
index 719cace..f5c111a 100644
--- a/samza-api/src/main/java/org/apache/samza/sql/udfs/ScalarUdf.java
+++ b/samza-api/src/main/java/org/apache/samza/sql/udfs/ScalarUdf.java
@@ -30,7 +30,7 @@ import org.apache.samza.config.Config;
* select myudf(id, name) from profile
* In the above query, Profile should contain fields named 'id' of INTEGER/NUMBER type and 'name' of type VARCHAR/CHARACTER
*/
-public interface ScalarUdf {
+public interface ScalarUdf<T> {
/**
* Udfs can implement this method to perform any initialization that they may need.
* @param udfConfig Config specific to the udf.
@@ -44,5 +44,5 @@ public interface ScalarUdf {
* @return
* Return value from the scalar udf.
*/
- Object execute(Object... args);
+ T execute(Object... args);
}
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java b/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java
index ab46a98..086bb93 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java
@@ -155,7 +155,7 @@ public class AvroRelConverter implements SamzaRelConverter {
.collect(Collectors.toList()));
} else if (value == null) {
fieldNames.addAll(relationalSchema.getFieldNames());
- IntStream.range(0, fieldNames.size() - 1).forEach(x -> values.add(null));
+ IntStream.range(0, fieldNames.size()).forEach(x -> values.add(null));
} else {
String msg = "Avro message converter doesn't support messages of type " + value.getClass();
LOG.error(msg);
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java b/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java
index b5df545..bf945a0 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java
@@ -23,6 +23,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
+import org.apache.commons.lang.Validate;
/**
@@ -51,6 +52,7 @@ public class SamzaSqlRelMessage {
* all the fields in the row can be null.
*/
public SamzaSqlRelMessage(Object key, List<String> names, List<Object> values) {
+ Validate.isTrue(names.size() == values.size(), "Field Names and values are not of same length.");
this.key = key;
this.value.addAll(values);
this.names.addAll(names);
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/main/java/org/apache/samza/sql/fn/FlattenUdf.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/fn/FlattenUdf.java b/samza-sql/src/main/java/org/apache/samza/sql/fn/FlattenUdf.java
index 39a26ce..e8d643d 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/fn/FlattenUdf.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/fn/FlattenUdf.java
@@ -24,7 +24,7 @@ import org.apache.samza.config.Config;
import org.apache.samza.sql.udfs.ScalarUdf;
-public class FlattenUdf implements ScalarUdf {
+public class FlattenUdf implements ScalarUdf<Object> {
@Override
public void init(Config udfConfig) {
}
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/main/java/org/apache/samza/sql/fn/RegexMatchUdf.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/fn/RegexMatchUdf.java b/samza-sql/src/main/java/org/apache/samza/sql/fn/RegexMatchUdf.java
new file mode 100644
index 0000000..1f5307d
--- /dev/null
+++ b/samza-sql/src/main/java/org/apache/samza/sql/fn/RegexMatchUdf.java
@@ -0,0 +1,39 @@
+/*
+* 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.samza.sql.fn;
+
+import java.util.regex.Pattern;
+import org.apache.samza.config.Config;
+import org.apache.samza.sql.udfs.ScalarUdf;
+
+
+/**
+ * Simple RegexMatch Udf.
+ */
+public class RegexMatchUdf implements ScalarUdf<Boolean> {
+ @Override
+ public void init(Config config) {
+
+ }
+
+ public Boolean execute(Object... args) {
+ return Pattern.matches((String) args[0], (String) args[1]);
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/main/java/org/apache/samza/sql/impl/ConfigBasedUdfResolver.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/impl/ConfigBasedUdfResolver.java b/samza-sql/src/main/java/org/apache/samza/sql/impl/ConfigBasedUdfResolver.java
index c98a4a1..412ff3b 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/impl/ConfigBasedUdfResolver.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/impl/ConfigBasedUdfResolver.java
@@ -27,7 +27,6 @@ import java.util.List;
import java.util.Optional;
import java.util.Properties;
import java.util.stream.Collectors;
-
import org.apache.commons.lang.StringUtils;
import org.apache.samza.SamzaException;
import org.apache.samza.config.Config;
@@ -74,19 +73,20 @@ public class ConfigBasedUdfResolver implements UdfResolver {
throw new SamzaException(msg);
}
- Optional<Method> udfMethod =
- Arrays.stream(udfClass.getMethods()).filter(x -> x.getName().equals(UDF_METHOD_NAME)).findFirst();
+ Method udfMethod;
- if (!udfMethod.isPresent()) {
+ try {
+ udfMethod = udfClass.getMethod(UDF_METHOD_NAME, Object[].class);
+ } catch (NoSuchMethodException e) {
String msg = String.format("Udf Class %s doesn't implement method named %s", udfClassName, UDF_METHOD_NAME);
LOG.error(msg);
- throw new SamzaException(msg);
+ throw new SamzaException(msg, e);
}
int udfIndex = udfClass.getSimpleName().toLowerCase().lastIndexOf("udf");
String udfName = udfClass.getSimpleName().substring(0, udfIndex);
- udfs.add(new UdfMetadata(udfName, udfMethod.get(), udfConfig.subset(udfName + ".")));
+ udfs.add(new UdfMetadata(udfName, udfMethod, udfConfig.subset(udfName + ".")));
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlScalarFunctionImpl.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlScalarFunctionImpl.java b/samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlScalarFunctionImpl.java
index 1200d57..6894c86 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlScalarFunctionImpl.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlScalarFunctionImpl.java
@@ -67,8 +67,11 @@ public class SamzaSqlScalarFunctionImpl implements ScalarFunction, Implementable
final Expression context = Expressions.parameter(SamzaSqlExecutionContext.class, "context");
final Expression getUdfInstance = Expressions.call(ScalarUdf.class, context, getUdfMethod,
Expressions.constant(udfMethod.getDeclaringClass().getName()), Expressions.constant(udfName));
- return Expressions.call(Expressions.convert_(getUdfInstance, udfMethod.getDeclaringClass()), udfMethod,
- translatedOperands);
+ final Expression callExpression = Expressions.convert_(Expressions.call(Expressions.convert_(getUdfInstance, udfMethod.getDeclaringClass()), udfMethod,
+ translatedOperands), Object.class);
+ // The Janino compiler which is used to compile the expressions doesn't seem to understand the Type of the ScalarUdf.execute
+ // because it is a generic. To work around that we are explicitly casting it to the return type.
+ return Expressions.convert_(callExpression, udfMethod.getReturnType());
}, NullPolicy.NONE, false);
}
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/test/java/org/apache/samza/sql/avro/TestAvroRelConversion.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/avro/TestAvroRelConversion.java b/samza-sql/src/test/java/org/apache/samza/sql/avro/TestAvroRelConversion.java
index a54db54..44b4213 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/avro/TestAvroRelConversion.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/avro/TestAvroRelConversion.java
@@ -137,6 +137,20 @@ public class TestAvroRelConversion {
LOG.info(Joiner.on(",").join(message.getFieldNames()));
}
+ @Test
+ public void testEmptyRecordConversion() {
+ GenericData.Record record = new GenericData.Record(SimpleRecord.SCHEMA$);
+ SamzaSqlRelMessage message = simpleRecordAvroRelConverter.convertToRelMessage(new KV<>("key", record));
+ Assert.assertEquals(message.getFieldNames().size(), message.getFieldValues().size());
+ }
+
+
+ @Test
+ public void testNullRecordConversion() {
+ SamzaSqlRelMessage message = simpleRecordAvroRelConverter.convertToRelMessage(new KV<>("key", null));
+ Assert.assertEquals(message.getFieldNames().size(), message.getFieldValues().size());
+ }
+
public static <T> byte[] encodeAvroSpecificRecord(Class<T> clazz, T record) throws IOException {
DatumWriter<T> msgDatumWriter = new SpecificDatumWriter<>(clazz);
ByteArrayOutputStream os = new ByteArrayOutputStream();
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/test/java/org/apache/samza/sql/e2e/TestSamzaSqlEndToEnd.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/e2e/TestSamzaSqlEndToEnd.java b/samza-sql/src/test/java/org/apache/samza/sql/e2e/TestSamzaSqlEndToEnd.java
index 769c653..8baa9e7 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/e2e/TestSamzaSqlEndToEnd.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/e2e/TestSamzaSqlEndToEnd.java
@@ -134,4 +134,20 @@ public class TestSamzaSqlEndToEnd {
Assert.assertTrue(
IntStream.range(0, numMessages).map(udf::execute).boxed().collect(Collectors.toList()).equals(outMessages));
}
+
+ @Test
+ public void testRegexMatchUdfInWhereClause() throws Exception {
+ int numMessages = 20;
+ TestAvroSystemFactory.messages.clear();
+ Map<String, String> staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(numMessages);
+ String sql1 = "Insert into testavro.outputTopic select id from testavro.SIMPLE1 where RegexMatch('.*4', Name)";
+ List<String> sqlStmts = Collections.singletonList(sql1);
+ staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts));
+ SamzaSqlApplicationRunner runner = new SamzaSqlApplicationRunner(true, new MapConfig(staticConfigs));
+ runner.runAndWaitForFinish();
+
+ LOG.info("output Messages " + TestAvroSystemFactory.messages);
+ // There should be two messages that contain "4"
+ Assert.assertEquals(TestAvroSystemFactory.messages.size(), 2);
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/test/java/org/apache/samza/sql/testutil/MyTestArrayUdf.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/testutil/MyTestArrayUdf.java b/samza-sql/src/test/java/org/apache/samza/sql/testutil/MyTestArrayUdf.java
index c9bc9bc..b2a2c25 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/testutil/MyTestArrayUdf.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/testutil/MyTestArrayUdf.java
@@ -19,18 +19,19 @@
package org.apache.samza.sql.testutil;
+import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.samza.config.Config;
import org.apache.samza.sql.udfs.ScalarUdf;
-public class MyTestArrayUdf implements ScalarUdf {
+public class MyTestArrayUdf implements ScalarUdf<List<String>> {
@Override
public void init(Config udfConfig) {
}
- public Object execute(Object... args) {
+ public List<String> execute(Object... args) {
Integer value = (Integer) args[0];
return IntStream.range(0, value).mapToObj(String::valueOf).collect(Collectors.toList());
}
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/test/java/org/apache/samza/sql/testutil/MyTestUdf.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/testutil/MyTestUdf.java b/samza-sql/src/test/java/org/apache/samza/sql/testutil/MyTestUdf.java
index baca367..c718368 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/testutil/MyTestUdf.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/testutil/MyTestUdf.java
@@ -28,7 +28,7 @@ import org.slf4j.LoggerFactory;
/**
* Test UDF used by unit and integration tests.
*/
-public class MyTestUdf implements ScalarUdf {
+public class MyTestUdf implements ScalarUdf<Integer> {
private static final Logger LOG = LoggerFactory.getLogger(MyTestUdf.class);
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-sql/src/test/java/org/apache/samza/sql/testutil/SamzaSqlTestConfig.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/testutil/SamzaSqlTestConfig.java b/samza-sql/src/test/java/org/apache/samza/sql/testutil/SamzaSqlTestConfig.java
index 760d5d0..008eb82 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/testutil/SamzaSqlTestConfig.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/testutil/SamzaSqlTestConfig.java
@@ -31,6 +31,7 @@ import org.apache.samza.sql.avro.ConfigBasedAvroRelSchemaProviderFactory;
import org.apache.samza.sql.avro.schemas.ComplexRecord;
import org.apache.samza.sql.avro.schemas.SimpleRecord;
import org.apache.samza.sql.fn.FlattenUdf;
+import org.apache.samza.sql.fn.RegexMatchUdf;
import org.apache.samza.sql.impl.ConfigBasedSourceResolverFactory;
import org.apache.samza.sql.impl.ConfigBasedUdfResolver;
import org.apache.samza.sql.interfaces.SqlSystemStreamConfig;
@@ -64,8 +65,9 @@ public class SamzaSqlTestConfig {
String configUdfResolverDomain = String.format(SamzaSqlApplicationConfig.CFG_FMT_UDF_RESOLVER_DOMAIN, "config");
staticConfigs.put(configUdfResolverDomain + SamzaSqlApplicationConfig.CFG_FACTORY,
ConfigBasedUdfResolver.class.getName());
- staticConfigs.put(configUdfResolverDomain + ConfigBasedUdfResolver.CFG_UDF_CLASSES,
- Joiner.on(",").join(MyTestUdf.class.getName(), FlattenUdf.class.getName(), MyTestArrayUdf.class.getName()));
+ staticConfigs.put(configUdfResolverDomain + ConfigBasedUdfResolver.CFG_UDF_CLASSES, Joiner.on(",")
+ .join(MyTestUdf.class.getName(), RegexMatchUdf.class.getName(), FlattenUdf.class.getName(),
+ MyTestArrayUdf.class.getName()));
String avroSystemConfigPrefix =
String.format(ConfigBasedSourceResolverFactory.CFG_FMT_SAMZA_PREFIX, SAMZA_SYSTEM_TEST_AVRO);
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-tools/src/main/java/org/apache/samza/tools/SamzaSqlConsole.java
----------------------------------------------------------------------
diff --git a/samza-tools/src/main/java/org/apache/samza/tools/SamzaSqlConsole.java b/samza-tools/src/main/java/org/apache/samza/tools/SamzaSqlConsole.java
index 9803117..11a49f7 100644
--- a/samza-tools/src/main/java/org/apache/samza/tools/SamzaSqlConsole.java
+++ b/samza-tools/src/main/java/org/apache/samza/tools/SamzaSqlConsole.java
@@ -20,12 +20,6 @@
package org.apache.samza.tools;
import com.google.common.base.Joiner;
-import org.apache.samza.tools.avro.AvroSchemaGenRelConverterFactory;
-import org.apache.samza.tools.avro.AvroSerDeFactory;
-import org.apache.samza.tools.json.JsonRelConverterFactory;
-import org.apache.samza.tools.schemas.PageViewEvent;
-import org.apache.samza.tools.schemas.ProfileChangeEvent;
-import org.apache.samza.tools.udf.RegexMatchUdf;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -43,6 +37,7 @@ import org.apache.samza.container.grouper.task.SingleContainerGrouperFactory;
import org.apache.samza.serializers.StringSerdeFactory;
import org.apache.samza.sql.avro.ConfigBasedAvroRelSchemaProviderFactory;
import org.apache.samza.sql.fn.FlattenUdf;
+import org.apache.samza.sql.fn.RegexMatchUdf;
import org.apache.samza.sql.impl.ConfigBasedSourceResolverFactory;
import org.apache.samza.sql.impl.ConfigBasedUdfResolver;
import org.apache.samza.sql.interfaces.SqlSystemStreamConfig;
@@ -52,6 +47,11 @@ import org.apache.samza.sql.testutil.JsonUtil;
import org.apache.samza.sql.testutil.SqlFileParser;
import org.apache.samza.standalone.PassthroughJobCoordinatorFactory;
import org.apache.samza.system.kafka.KafkaSystemFactory;
+import org.apache.samza.tools.avro.AvroSchemaGenRelConverterFactory;
+import org.apache.samza.tools.avro.AvroSerDeFactory;
+import org.apache.samza.tools.json.JsonRelConverterFactory;
+import org.apache.samza.tools.schemas.PageViewEvent;
+import org.apache.samza.tools.schemas.ProfileChangeEvent;
public class SamzaSqlConsole {
http://git-wip-us.apache.org/repos/asf/samza/blob/9893069c/samza-tools/src/main/java/org/apache/samza/tools/udf/RegexMatchUdf.java
----------------------------------------------------------------------
diff --git a/samza-tools/src/main/java/org/apache/samza/tools/udf/RegexMatchUdf.java b/samza-tools/src/main/java/org/apache/samza/tools/udf/RegexMatchUdf.java
deleted file mode 100644
index 42e9f04..0000000
--- a/samza-tools/src/main/java/org/apache/samza/tools/udf/RegexMatchUdf.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
-* 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.samza.tools.udf;
-
-import java.util.regex.Pattern;
-import org.apache.samza.config.Config;
-import org.apache.samza.sql.udfs.ScalarUdf;
-
-
-/**
- * Simple RegexMatch Udf.
- */
-public class RegexMatchUdf implements ScalarUdf {
- @Override
- public void init(Config config) {
-
- }
-
- @Override
- public Object execute(Object... args) {
- return Pattern.matches((String) args[0], (String) args[1]);
- }
-}