You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hawq.apache.org by ad...@apache.org on 2018/06/01 19:32:38 UTC
[1/2] incubator-hawq git commit: HAWQ-1599. Add PXF-Ignite plug-in.
Repository: incubator-hawq
Updated Branches:
refs/heads/master e56a28158 -> 3779f14dc
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgnitePartitionFragmenterTest.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgnitePartitionFragmenterTest.java b/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgnitePartitionFragmenterTest.java
new file mode 100644
index 0000000..0dd0720
--- /dev/null
+++ b/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgnitePartitionFragmenterTest.java
@@ -0,0 +1,259 @@
+package org.apache.hawq.pxf.plugins.ignite;
+
+/*
+ * 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.
+ */
+
+import org.apache.hawq.pxf.api.Fragment;
+import org.apache.hawq.pxf.api.UserDataException;
+import org.apache.hawq.pxf.api.utilities.InputData;
+
+import java.util.Calendar;
+import java.util.List;
+
+import org.apache.commons.compress.utils.ByteUtils;
+
+import org.junit.Before;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+public class IgnitePartitionFragmenterTest {
+ @Before
+ public void preparePartitionFragmenterTest() throws Exception {
+ inputData = mock(InputData.class);
+ when(inputData.getDataSource()).thenReturn("sales");
+ }
+
+ @Test
+ public void testPartitionByDateOfMonth() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("cdate:date");
+ when(inputData.getUserProperty("RANGE")).thenReturn("2008-01-01:2009-01-01");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("1:month");
+
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+ assertEquals(12, fragments.size());
+
+ // Fragment 1
+ byte[] fragMeta = fragments.get(0).getMetadata();
+ long fragStart = ByteUtils.fromLittleEndian(fragMeta, 0, 8);
+ long fragEnd = ByteUtils.fromLittleEndian(fragMeta, 8, 8);
+ assertDateEquals(fragStart, 2008, 1, 1);
+ assertDateEquals(fragEnd, 2008, 2, 1);
+
+ // Fragment 12
+ fragMeta = fragments.get(11).getMetadata();
+ fragStart = ByteUtils.fromLittleEndian(fragMeta, 0, 8);
+ fragEnd = ByteUtils.fromLittleEndian(fragMeta, 8, 8);
+ assertDateEquals(fragStart, 2008, 12, 1);
+ assertDateEquals(fragEnd, 2009, 1, 1);
+
+ // End date > Start date
+ when(inputData.getUserProperty("RANGE")).thenReturn("2008-01-01:2001-01-01");
+ fragment = new IgnitePartitionFragmenter(inputData);
+ fragments = fragment.getFragments();
+ assertEquals(0, fragments.size());
+ }
+
+ @Test
+ public void testPartitionByDateOfYear() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("cdate:date");
+ when(inputData.getUserProperty("RANGE")).thenReturn("2008-01-01:2011-01-01");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("1:year");
+
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+ assertEquals(3, fragments.size());
+ }
+
+ @Test
+ public void testPartitionByInt() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("year:int");
+ when(inputData.getUserProperty("RANGE")).thenReturn("2001:2012");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("2");
+
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+ assertEquals(6, fragments.size());
+
+ // Fragment 1
+ byte[] fragMeta = fragments.get(0).getMetadata();
+ int fragStart = (int)ByteUtils.fromLittleEndian(fragMeta, 0, 4);
+ int fragEnd = (int)ByteUtils.fromLittleEndian(fragMeta, 4, 4);
+ assertEquals(2001, fragStart);
+ assertEquals(2003, fragEnd);
+
+ // Fragment 6
+ fragMeta = fragments.get(5).getMetadata();
+ fragStart = (int)ByteUtils.fromLittleEndian(fragMeta, 0, 4);
+ fragEnd = (int)ByteUtils.fromLittleEndian(fragMeta, 4, 4);
+ assertEquals(2011, fragStart);
+ assertEquals(2012, fragEnd);
+
+ // End > Start
+ when(inputData.getUserProperty("RANGE")).thenReturn("2013:2012");
+ fragment = new IgnitePartitionFragmenter(inputData);
+ assertEquals(0, fragment.getFragments().size());
+ }
+
+ @Test
+ public void testPartitionByEnum() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("level:enum");
+ when(inputData.getUserProperty("RANGE")).thenReturn("excellent:good:general:bad");
+
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+ assertEquals(4, fragments.size());
+
+ //fragment - 1
+ byte[] fragMeta = fragments.get(0).getMetadata();
+ assertEquals("excellent", new String(fragMeta));
+
+ //fragment - 4
+ fragMeta = fragments.get(3).getMetadata();
+ assertEquals("bad", new String(fragMeta));
+ }
+
+ @Test(expected = UserDataException.class)
+ public void testInvalidPartitiontype() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("level:float");
+ when(inputData.getUserProperty("RANGE")).thenReturn("100:200");
+
+ new IgnitePartitionFragmenter(inputData);
+ }
+
+ @Test(expected = UserDataException.class)
+ public void testInvalidParameterFormat() throws Exception {
+ //PARTITION_BY must be comma-delimited string
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("level-enum");
+ when(inputData.getUserProperty("RANGE")).thenReturn("100:200");
+
+ new IgnitePartitionFragmenter(inputData);
+ }
+
+ @Test(expected = UserDataException.class)
+ public void testInvalidDateFormat() throws Exception {
+ //date string must be yyyy-MM-dd
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("cdate:date");
+ when(inputData.getUserProperty("RANGE")).thenReturn("2008/01/01:2009-01-01");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("1:month");
+
+ new IgnitePartitionFragmenter(inputData).getFragments();
+ }
+
+ @Test(expected = UserDataException.class)
+ public void testInvalidParameterValue() throws Exception {
+ //INTERVAL must be greater than 0
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("cdate:date");
+ when(inputData.getUserProperty("RANGE")).thenReturn("2008-01-01:2009-01-01");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("-1:month");
+
+ new IgnitePartitionFragmenter(inputData);
+ }
+
+ @Test(expected = UserDataException.class)
+ public void testInvalidIntervalType() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("cdate:date");
+ when(inputData.getUserProperty("RANGE")).thenReturn("2008-01-01:2011-01-01");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("6:hour");
+
+ new IgnitePartitionFragmenter(inputData).getFragments();
+ }
+
+ @Test(expected = UserDataException.class)
+ public void testIntervalTypeMissing() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("cdate:date");
+ when(inputData.getUserProperty("RANGE")).thenReturn("2008-01-01:2011-01-01");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("6");
+
+ new IgnitePartitionFragmenter(inputData).getFragments();
+ }
+
+ @Test
+ public void testIntervalTypeMissingValid() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("year:int");
+ when(inputData.getUserProperty("RANGE")).thenReturn("2001:2012");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("1");
+
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+ assertEquals(11, fragments.size());
+ }
+
+ @Test
+ public void testIntervalMissingEnum() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("level:enum");
+ when(inputData.getUserProperty("RANGE")).thenReturn("100:200:300");
+
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+ assertEquals(3, fragments.size());
+ }
+
+ @Test(expected = UserDataException.class)
+ public void testRangeMissingEndValue() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("cdate:date");
+ when(inputData.getUserProperty("RANGE")).thenReturn("2008-01-01");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("1:year");
+
+ new IgnitePartitionFragmenter(inputData).getFragments();
+ }
+
+ @Test(expected = UserDataException.class)
+ public void testRangeMissing() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("year:int");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("1");
+
+ new IgnitePartitionFragmenter(inputData).getFragments();
+ }
+
+ @Test
+ public void testRangeSingleValueEnum() throws Exception {
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("level:enum");
+ when(inputData.getUserProperty("RANGE")).thenReturn("100");
+
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+ assertEquals(1, fragments.size());
+ }
+
+ @Test
+ public void testNoPartition() throws Exception {
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+ assertEquals(1, fragments.size());
+ }
+
+
+ private InputData inputData = null;
+
+ /**
+ * Check if two dates are equal and throw an exception if they are not
+ */
+ private void assertDateEquals(long date, int year, int month, int day) {
+ Calendar calendar = Calendar.getInstance();
+ calendar.setTimeInMillis(date);
+ assertEquals(calendar.get(Calendar.YEAR), year);
+ assertEquals(calendar.get(Calendar.MONTH), month - 1);
+ assertEquals(calendar.get(Calendar.DAY_OF_MONTH), day);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/SqlBuilderTest.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/SqlBuilderTest.java b/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/SqlBuilderTest.java
new file mode 100644
index 0000000..0b3ebf4
--- /dev/null
+++ b/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/SqlBuilderTest.java
@@ -0,0 +1,162 @@
+package org.apache.hawq.pxf.plugins.ignite;
+
+/*
+ * 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.
+ */
+
+import org.apache.hawq.pxf.api.*;
+import org.apache.hawq.pxf.api.utilities.ColumnDescriptor;
+import org.apache.hawq.pxf.api.utilities.InputData;
+import org.apache.hawq.pxf.api.io.DataType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+import static org.junit.Assert.assertEquals;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+/**
+ * Validate SQL string generated by the {@link IgnitePartitionFragmenter#buildFragmenterSql} method
+ * and the {@link WhereSQLBuilder#buildWhereSQL} method.
+ */
+public class SqlBuilderTest {
+ @Test
+ public void testIdFilter() throws Exception {
+ prepareConstruction();
+ when(inputData.hasFilter()).thenReturn(true);
+ when(inputData.getFilterString()).thenReturn("a0c20s1d1o5"); // id=1
+
+ WhereSQLBuilder builder = new WhereSQLBuilder(inputData);
+ assertEquals("id=1", builder.buildWhereSQL());
+ }
+
+ @Test
+ public void testDateAndAmtFilter() throws Exception {
+ prepareConstruction();
+ when(inputData.hasFilter()).thenReturn(true);
+ // cdate>'2008-02-01' and cdate<'2008-12-01' and amt > 1200
+ when(inputData.getFilterString()).thenReturn("a1c25s10d2008-02-01o2a1c25s10d2008-12-01o1l0a2c20s4d1200o2l0");
+
+ WhereSQLBuilder builder = new WhereSQLBuilder(inputData);
+ assertEquals("cdate>'2008-02-01' AND cdate<'2008-12-01' AND amt>1200"
+ , builder.buildWhereSQL());
+ }
+
+ @Test
+ public void testUnsupportedOperationFilter() throws Exception {
+ prepareConstruction();
+ when(inputData.hasFilter()).thenReturn(true);
+ // grade like 'bad'
+ when(inputData.getFilterString()).thenReturn("a3c25s3dbado7");
+
+ WhereSQLBuilder builder = new WhereSQLBuilder(inputData);
+ assertEquals(null, builder.buildWhereSQL());
+ }
+
+ @Test
+ public void testUnsupportedLogicalFilter() throws Exception {
+ prepareConstruction();
+ when(inputData.hasFilter()).thenReturn(true);
+ // cdate>'2008-02-01' or amt < 1200
+ when(inputData.getFilterString()).thenReturn("a1c25s10d2008-02-01o2a2c20s4d1200o2l1");
+
+ WhereSQLBuilder builder = new WhereSQLBuilder(inputData);
+ assertEquals(null, builder.buildWhereSQL());
+ }
+
+ @Test
+ public void testDatePartition() throws Exception {
+ prepareConstruction();
+ when(inputData.hasFilter()).thenReturn(false);
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("cdate:date");
+ when(inputData.getUserProperty("RANGE")).thenReturn("2008-01-01:2009-01-01");
+ when(inputData.getUserProperty("INTERVAL")).thenReturn("2:month");
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+ assertEquals(6, fragments.size());
+
+ // Partition 1: cdate>=2008-01-01 and cdate<2008-03-01
+ when(inputData.getFragmentMetadata()).thenReturn(fragments.get(0).getMetadata());
+ StringBuilder sb = new StringBuilder(ORIGINAL_SQL);
+ IgnitePartitionFragmenter.buildFragmenterSql(inputData, sb);
+ assertEquals(ORIGINAL_SQL + " WHERE cdate>='2008-01-01' AND cdate<'2008-03-01'", sb.toString());
+ }
+
+ @Test
+ public void testFilterAndPartition() throws Exception {
+ prepareConstruction();
+ when(inputData.hasFilter()).thenReturn(true);
+ when(inputData.getFilterString()).thenReturn("a0c20s1d5o2"); //id>5
+ when(inputData.getUserProperty("PARTITION_BY")).thenReturn("grade:enum");
+ when(inputData.getUserProperty("RANGE")).thenReturn("excellent:good:general:bad");
+
+ WhereSQLBuilder builder = new WhereSQLBuilder(inputData);
+ String whereSql = builder.buildWhereSQL();
+ assertEquals("id>5", whereSql);
+
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+
+ // Partition 1: id>5 and grade='excellent'
+ when(inputData.getFragmentMetadata()).thenReturn(fragments.get(0).getMetadata());
+
+ String filterSql = ORIGINAL_SQL + " WHERE " + whereSql;
+ StringBuilder sb = new StringBuilder(filterSql);
+ IgnitePartitionFragmenter.buildFragmenterSql(inputData, sb);
+ assertEquals(filterSql + " AND grade='excellent'", sb.toString());
+ }
+
+ @Test
+ public void testNoPartition() throws Exception {
+ prepareConstruction();
+ when(inputData.hasFilter()).thenReturn(false);
+ IgnitePartitionFragmenter fragment = new IgnitePartitionFragmenter(inputData);
+ List<Fragment> fragments = fragment.getFragments();
+ assertEquals(1, fragments.size());
+
+ when(inputData.getFragmentMetadata()).thenReturn(fragments.get(0).getMetadata());
+
+ StringBuilder sb = new StringBuilder(ORIGINAL_SQL);
+ IgnitePartitionFragmenter.buildFragmenterSql(inputData, sb);
+ assertEquals(ORIGINAL_SQL, sb.toString());
+ }
+
+ private void prepareConstruction() throws Exception {
+ inputData = mock(InputData.class);
+ when(inputData.getDataSource()).thenReturn("sales");
+
+ ArrayList<ColumnDescriptor> columns = new ArrayList<>();
+ columns.add(new ColumnDescriptor("id", DataType.INTEGER.getOID(), 0, "int4", null));
+ columns.add(new ColumnDescriptor("cdate", DataType.DATE.getOID(), 1, "date", null));
+ columns.add(new ColumnDescriptor("amt", DataType.FLOAT8.getOID(), 2, "float8", null));
+ columns.add(new ColumnDescriptor("grade", DataType.TEXT.getOID(), 3, "text", null));
+ when(inputData.getTupleDescription()).thenReturn(columns);
+ when(inputData.getColumn(0)).thenReturn(columns.get(0));
+ when(inputData.getColumn(1)).thenReturn(columns.get(1));
+ when(inputData.getColumn(2)).thenReturn(columns.get(2));
+ when(inputData.getColumn(3)).thenReturn(columns.get(3));
+ }
+
+
+ private static final String ORIGINAL_SQL = "SELECT * FROM sales";
+ private InputData inputData;
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-service/src/configs/templates/pxf-private-cdh.classpath.template
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/configs/templates/pxf-private-cdh.classpath.template b/pxf/pxf-service/src/configs/templates/pxf-private-cdh.classpath.template
index 437ffaf..7a54316 100644
--- a/pxf/pxf-service/src/configs/templates/pxf-private-cdh.classpath.template
+++ b/pxf/pxf-service/src/configs/templates/pxf-private-cdh.classpath.template
@@ -18,7 +18,7 @@
# with RPM-based installation of hadoop clients from CDH distribution.
# WARNING: DO NOT EDIT!
# Any change in this file can result in PXF failing to run.
-# Adding resources should be done using pxf-public.classpath file.
+# Adding resources should be done using pxf-public.classpath file.
##################################################################
# PXF Configuration
@@ -30,6 +30,7 @@ PXF_HOME/lib/pxf-hdfs.jar
PXF_HOME/lib/pxf-hive.jar
PXF_HOME/lib/pxf-json.jar
PXF_HOME/lib/pxf-jdbc.jar
+PXF_HOME/lib/pxf-ignite.jar
# Hadoop/Hive/HBase configurations
/etc/hadoop/conf
@@ -50,6 +51,7 @@ PXF_HOME/lib/pxf-jdbc.jar
/usr/lib/hadoop/client/commons-io.jar
/usr/lib/hadoop/client/commons-lang.jar
/usr/lib/hadoop/client/commons-logging.jar
+/usr/lib/hadoop/client/commons-compress.jar
/usr/lib/hadoop/client/guava.jar
/usr/lib/hadoop/client/htrace-core4.jar
/usr/lib/hadoop/client/jackson-core.jar
@@ -62,6 +64,7 @@ PXF_HOME/lib/pxf-jdbc.jar
/usr/lib/hadoop/client/snappy-java.jar
/usr/lib/hadoop/lib/asm-*[0-9].jar
/usr/lib/hadoop/lib/jersey-server-*[0-9].jar
+/usr/lib/hadoop/client/gson-*[0-9].jar
# Hive Libraries
/usr/lib/hive/lib/antlr-runtime-*[0-9].jar
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-service/src/configs/templates/pxf-private-hdp.classpath.template
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/configs/templates/pxf-private-hdp.classpath.template b/pxf/pxf-service/src/configs/templates/pxf-private-hdp.classpath.template
index da834a6..7e24cec 100644
--- a/pxf/pxf-service/src/configs/templates/pxf-private-hdp.classpath.template
+++ b/pxf/pxf-service/src/configs/templates/pxf-private-hdp.classpath.template
@@ -18,7 +18,7 @@
# with RPM-based installation of hadoop clients from HDP distribution.
# WARNING: DO NOT EDIT!
# Any change in this file can result in PXF failing to run.
-# Adding resources should be done using pxf-public.classpath file.
+# Adding resources should be done using pxf-public.classpath file.
##################################################################
# PXF Configuration
@@ -30,6 +30,7 @@ PXF_HOME/lib/pxf-hdfs.jar
PXF_HOME/lib/pxf-hive.jar
PXF_HOME/lib/pxf-json.jar
PXF_HOME/lib/pxf-jdbc.jar
+PXF_HOME/lib/pxf-ignite.jar
# Hadoop/Hive/HBase configurations
/etc/hadoop/conf
@@ -50,6 +51,7 @@ PXF_HOME/lib/pxf-jdbc.jar
/usr/hdp/current/hadoop-client/client/commons-io.jar
/usr/hdp/current/hadoop-client/client/commons-lang.jar
/usr/hdp/current/hadoop-client/client/commons-logging.jar
+/usr/hdp/current/hadoop-client/client/commons-compress.jar
/usr/hdp/current/hadoop-client/client/guava.jar
/usr/hdp/current/hadoop-client/client/htrace-core.jar
/usr/hdp/current/hadoop-client/client/jackson-core-asl.jar
@@ -62,6 +64,7 @@ PXF_HOME/lib/pxf-jdbc.jar
/usr/hdp/current/hadoop-client/client/snappy-java.jar
/usr/hdp/current/hadoop-client/lib/asm-*[0-9].jar
/usr/hdp/current/hadoop-client/lib/jersey-server-*[0-9].jar
+/usr/hdp/current/hadoop-client/client/gson-*[0-9].jar
# Hive Libraries
/usr/hdp/current/hive-client/lib/antlr-runtime-*[0-9].jar
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-service/src/configs/templates/pxf-private-tar.classpath.template
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/configs/templates/pxf-private-tar.classpath.template b/pxf/pxf-service/src/configs/templates/pxf-private-tar.classpath.template
index 24c5cf2..d046a37 100644
--- a/pxf/pxf-service/src/configs/templates/pxf-private-tar.classpath.template
+++ b/pxf/pxf-service/src/configs/templates/pxf-private-tar.classpath.template
@@ -29,6 +29,7 @@ PXF_HOME/lib/pxf-hdfs.jar
PXF_HOME/lib/pxf-hive.jar
PXF_HOME/lib/pxf-json.jar
PXF_HOME/lib/pxf-jdbc.jar
+PXF_HOME/lib/pxf-ignite.jar
# Hadoop/Hive/HBase configurations
HADOOP_ROOT/hadoop/etc/hadoop
@@ -47,6 +48,7 @@ HADOOP_ROOT/hadoop/share/hadoop/common/lib/commons-configuration-*[0-9].jar
HADOOP_ROOT/hadoop/share/hadoop/common/lib/commons-io-*[0-9].jar
HADOOP_ROOT/hadoop/share/hadoop/common/lib/commons-lang-*[0-9].jar
HADOOP_ROOT/hadoop/share/hadoop/common/lib/commons-logging-*[0-9].jar
+HADOOP_ROOT/hadoop/share/hadoop/common/lib/commons-compress-*[0-9].jar
HADOOP_ROOT/hadoop/share/hadoop/common/lib/guava-*[0-9].jar
HADOOP_ROOT/hadoop/share/hadoop/common/lib/htrace-core*.jar
HADOOP_ROOT/hadoop/share/hadoop/common/lib/jetty-*.jar
@@ -59,6 +61,7 @@ HADOOP_ROOT/hadoop/share/hadoop/common/lib/snappy-java-*[0-9].jar
HADOOP_ROOT/hadoop/share/hadoop/hdfs/hadoop-hdfs-*[0-9].jar
HADOOP_ROOT/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-core-*[0-9].jar
HADOOP_ROOT/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-common-*[0-9].jar
+HADOOP_ROOT/hadoop/share/hadoop/common/lib/gson-*[0-9].jar
# Pick Jackson 1.9 jars from hdfs dir for HDP tar and from mapreduce1 for CDH tar
HADOOP_ROOT/hadoop/share/hadoop/[hdfs|madreduce1]/lib/jackson-core-asl-1.9*[0-9].jar
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-service/src/main/resources/pxf-private.classpath
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/resources/pxf-private.classpath b/pxf/pxf-service/src/main/resources/pxf-private.classpath
index 4ad2762..4a5716f 100644
--- a/pxf/pxf-service/src/main/resources/pxf-private.classpath
+++ b/pxf/pxf-service/src/main/resources/pxf-private.classpath
@@ -37,6 +37,7 @@ pxf/pxf-hdfs-*[0-9].jar
pxf/pxf-hive-*[0-9].jar
pxf/pxf-json-*[0-9].jar
pxf/pxf-jdbc-*[0-9].jar
+pxf/pxf-ignite-*[0-9].jar
# Hadoop Libraries
hadoop/share/hadoop/hdfs/hadoop-hdfs-*[0-9].jar
@@ -52,6 +53,7 @@ hadoop/share/hadoop/common/lib/commons-configuration-*[0-9].jar
hadoop/share/hadoop/common/lib/commons-io-*[0-9].jar
hadoop/share/hadoop/common/lib/commons-lang-*[0-9].jar
hadoop/share/hadoop/common/lib/commons-logging-*[0-9].jar
+hadoop/share/hadoop/common/lib/commons-compress-*[0-9].jar
hadoop/share/hadoop/common/lib/guava-*[0-9].jar
hadoop/share/hadoop/common/lib/htrace-core*.jar
hadoop/share/hadoop/common/lib/jetty-*.jar
@@ -62,6 +64,7 @@ hadoop/share/hadoop/common/lib/jersey-server-*[0-9].jar
hadoop/share/hadoop/common/lib/log4j-*[0-9].jar
hadoop/share/hadoop/common/lib/protobuf-java-*[0-9].jar
hadoop/share/hadoop/common/lib/slf4j-api-*[0-9].jar
+hadoop/share/hadoop/common/lib/gson-*[0-9].jar
# Hive Libraries
hive/lib/antlr-runtime*.jar
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-service/src/main/resources/pxf-privatebigtop.classpath
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/resources/pxf-privatebigtop.classpath b/pxf/pxf-service/src/main/resources/pxf-privatebigtop.classpath
index e949dcc..abfef15 100644
--- a/pxf/pxf-service/src/main/resources/pxf-privatebigtop.classpath
+++ b/pxf/pxf-service/src/main/resources/pxf-privatebigtop.classpath
@@ -17,7 +17,7 @@
# This file contains the internal classpaths required to run PXF.
# WARNING: DO NOT EDIT!
# Any change in this file can result in PXF failing to run.
-# Adding resources should be done using pxf-public.classpath file.
+# Adding resources should be done using pxf-public.classpath file.
##################################################################
/etc/pxf/conf
/etc/hadoop/conf
@@ -39,6 +39,7 @@
/usr/lib/hadoop/client/commons-io-*[0-9].jar
/usr/lib/hadoop/client/commons-lang-*[0-9].jar
/usr/lib/hadoop/client/commons-logging-*[0-9].jar
+/usr/lib/hadoop/client/commons-compress-*[0-9].jar
/usr/lib/hadoop/client/guava-*[0-9].jar
/usr/lib/hadoop/client/htrace-core-*[0-9]*.jar
/usr/lib/hadoop/client/hadoop-lzo-*[0-9].jar
@@ -49,6 +50,7 @@
/usr/lib/hadoop/client/protobuf-java-*[0-9].jar
/usr/lib/hadoop/client/slf4j-api-*[0-9].jar
/usr/lib/hadoop/client/snappy-java-*[0-9].jar
+/usr/lib/hadoop/client/gson-*[0-9].jar
/usr/lib/hive/lib/antlr-runtime-*[0-9].jar
/usr/lib/hive/lib/datanucleus-api-jdo-*[0-9].jar
/usr/lib/hive/lib/datanucleus-core-*[0-9].jar
@@ -61,3 +63,4 @@
/usr/lib/pxf/pxf-hive-*[0-9].jar
/usr/lib/pxf/pxf-json-*[0-9].jar
/usr/lib/pxf/pxf-jdbc-*[0-9].jar
+/usr/lib/pxf/pxf-ignite-*[0-9].jar
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-service/src/main/resources/pxf-privatehdp.classpath
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/resources/pxf-privatehdp.classpath b/pxf/pxf-service/src/main/resources/pxf-privatehdp.classpath
index 89e7ce2..8a445ca 100644
--- a/pxf/pxf-service/src/main/resources/pxf-privatehdp.classpath
+++ b/pxf/pxf-service/src/main/resources/pxf-privatehdp.classpath
@@ -16,7 +16,7 @@
# This file contains the internal classpaths required to run PXF.
# WARNING: DO NOT EDIT!
# Any change in this file can result in PXF failing to run.
-# Adding resources should be done using pxf-public.classpath file.
+# Adding resources should be done using pxf-public.classpath file.
##################################################################
/etc/pxf/conf
/etc/hadoop/conf
@@ -35,6 +35,7 @@
/usr/hdp/current/hadoop-client/lib/commons-io-*[0-9].jar
/usr/hdp/current/hadoop-client/lib/commons-lang-*[0-9].jar
/usr/hdp/current/hadoop-client/lib/commons-logging-*[0-9].jar
+/usr/hdp/current/hadoop-client/lib/commons-compress-*[0-9].jar
/usr/hdp/current/hadoop-client/lib/guava-*[0-9].jar
/usr/hdp/current/hadoop-client/lib/htrace-core-*[0-9]*.jar
/usr/hdp/current/hadoop-client/lib/hadoop-lzo-*[0-9].jar
@@ -47,6 +48,7 @@
/usr/hdp/current/hadoop-client/lib/protobuf-java-*[0-9].jar
/usr/hdp/current/hadoop-client/lib/slf4j-api-*[0-9].jar
/usr/hdp/current/hadoop-client/lib/snappy-java-*[0-9].jar
+/usr/hdp/current/hadoop-client/lib/gson-*[0-9].jar
/usr/hdp/current/hbase-client/lib/hbase-client.jar
/usr/hdp/current/hbase-client/lib/hbase-common.jar
/usr/hdp/current/hbase-client/lib/hbase-protocol.jar
@@ -66,3 +68,4 @@
/usr/lib/pxf/pxf-hive-*[0-9].jar
/usr/lib/pxf/pxf-json-*[0-9].jar
/usr/lib/pxf/pxf-jdbc-*[0-9].jar
+/usr/lib/pxf/pxf-ignite-*[0-9].jar
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-service/src/main/resources/pxf-privatephd.classpath
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/resources/pxf-privatephd.classpath b/pxf/pxf-service/src/main/resources/pxf-privatephd.classpath
index a380033..5319acb 100644
--- a/pxf/pxf-service/src/main/resources/pxf-privatephd.classpath
+++ b/pxf/pxf-service/src/main/resources/pxf-privatephd.classpath
@@ -16,7 +16,7 @@
# This file contains the internal classpaths required to run PXF.
# WARNING: DO NOT EDIT!
# Any change in this file can result in PXF failing to run.
-# Adding resources should be done using pxf-public.classpath file.
+# Adding resources should be done using pxf-public.classpath file.
##################################################################
/etc/pxf/conf
/etc/hadoop/conf
@@ -35,6 +35,7 @@
/usr/phd/current/hadoop-client/lib/commons-io-*[0-9].jar
/usr/phd/current/hadoop-client/lib/commons-lang-*[0-9].jar
/usr/phd/current/hadoop-client/lib/commons-logging-*[0-9].jar
+/usr/phd/current/hadoop-client/lib/commons-compress-*[0-9].jar
/usr/phd/current/hadoop-client/lib/guava-*[0-9].jar
/usr/phd/current/hadoop-client/lib/htrace-core-*[0-9]*.jar
/usr/phd/current/hadoop-client/lib/hadoop-lzo-*[0-9].jar
@@ -47,6 +48,7 @@
/usr/phd/current/hadoop-client/lib/protobuf-java-*[0-9].jar
/usr/phd/current/hadoop-client/lib/slf4j-api-*[0-9].jar
/usr/phd/current/hadoop-client/lib/snappy-java-*[0-9].jar
+/usr/phd/current/hadoop-client/lib/gson-*[0-9].jar
/usr/phd/current/hbase-client/lib/hbase-client.jar
/usr/phd/current/hbase-client/lib/hbase-common.jar
/usr/phd/current/hbase-client/lib/hbase-protocol.jar
@@ -65,3 +67,4 @@
/usr/lib/pxf/pxf-hive-*[0-9].jar
/usr/lib/pxf/pxf-json-*[0-9].jar
/usr/lib/pxf/pxf-jdbc-*[0-9].jar
+/usr/lib/pxf/pxf-ignite-*[0-9].jar
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-service/src/main/resources/pxf-profiles-default.xml
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/resources/pxf-profiles-default.xml b/pxf/pxf-service/src/main/resources/pxf-profiles-default.xml
index 0a4c2f4..252791e 100644
--- a/pxf/pxf-service/src/main/resources/pxf-profiles-default.xml
+++ b/pxf/pxf-service/src/main/resources/pxf-profiles-default.xml
@@ -193,4 +193,13 @@ under the License.
<resolver>org.apache.hawq.pxf.plugins.hdfs.ParquetResolver</resolver>
</plugins>
</profile>
+ <profile>
+ <name>Apache Ignite</name>
+ <description>A profile to read and write data from/to Apache Ignite database</description>
+ <plugins>
+ <fragmenter>org.apache.hawq.pxf.plugins.ignite.IgnitePartitionFragmenter</fragmenter>
+ <accessor>org.apache.hawq.pxf.plugins.ignite.IgniteAccessor</accessor>
+ <resolver>org.apache.hawq.pxf.plugins.ignite.IgniteResolver</resolver>
+ </plugins>
+ </profile>
</profiles>
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-service/src/main/resources/pxf-profiles.xml
----------------------------------------------------------------------
diff --git a/pxf/pxf-service/src/main/resources/pxf-profiles.xml b/pxf/pxf-service/src/main/resources/pxf-profiles.xml
index 796f49b..3fa651d 100644
--- a/pxf/pxf-service/src/main/resources/pxf-profiles.xml
+++ b/pxf/pxf-service/src/main/resources/pxf-profiles.xml
@@ -33,4 +33,4 @@ under the License.
</profile>
-->
<profiles>
-</profiles>
\ No newline at end of file
+</profiles>
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/settings.gradle
----------------------------------------------------------------------
diff --git a/pxf/settings.gradle b/pxf/settings.gradle
index e26f812..fa60ea4 100644
--- a/pxf/settings.gradle
+++ b/pxf/settings.gradle
@@ -27,3 +27,4 @@ include 'pxf-hive'
include 'pxf-hbase'
include 'pxf-json'
include 'pxf-jdbc'
+include 'pxf-ignite'
[2/2] incubator-hawq git commit: HAWQ-1599. Add PXF-Ignite plug-in.
Posted by ad...@apache.org.
HAWQ-1599. Add PXF-Ignite plug-in.
(close #1344)
The PXF-Ignite plug-in allows to read data from Ignite database.
The syntax is similar to one of PXF-JDBC plug-in.
This implementation uses Ignite REST API. It can be easily modified to access any other databases or services via REST API.
This commit does not change `pxf-service` links and classpath files. They should be added manually in order for the PXF-Ignite to work.
See pxf-ignite/README.md for instructions on how to use this plug-in.
PXF Ignite plug-in: Implement write access
This commit adds the support of write operations to the PXF Ignite plug-in.
PXF Ignite plug-in: tests and fixes
PXF Ignite plug-in: fixes proposed by contributors
PXF Ignite plug-in: Fixes proposed by @hornn, @denalex
PXF Ignite plug-in: Fixed nullptr exception and invalid bufferWrite operations
PXF Ignite plug-in: Fixes proposed by @denalex, @sansanichfb
Use 'apache.commons.compress.utils.ByteUtils' instead of a custom 'ByteUtil' library to process fragment metadata;
This also removes the dependence of 'apache.commons.lang.ArrayUtils'.
This fix was proposed by @denalex.
Change signature of the 'buildFragmenterSql()' function in 'IgnitePartitionFragmenter', so that it returns void.
This fix was proposed by @sansanichfb.
PXF Ignite plug-in: minor codestyle improvements
PXF Ignite plug-in: README.md update
PXF Ignite plug-in: Fix handling of TIMESTAMP
Fix handling of TIMESTAMP values with milli- and microseconds
SimpleDateFormat objects are now ThreadLocal (due to safety reasons)
Project: http://git-wip-us.apache.org/repos/asf/incubator-hawq/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-hawq/commit/3779f14d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-hawq/tree/3779f14d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-hawq/diff/3779f14d
Branch: refs/heads/master
Commit: 3779f14dc62626a6afefa4aa4322b450d0b09da1
Parents: e56a281
Author: Ivan Leskin <le...@arenadata.io>
Authored: Thu Feb 8 16:11:19 2018 +0300
Committer: Alexander Denissov <ad...@pivotal.io>
Committed: Fri Jun 1 12:31:19 2018 -0700
----------------------------------------------------------------------
pxf/build.gradle | 32 +-
pxf/pxf-ignite/README.md | 79 +++
.../hawq/pxf/plugins/ignite/IgniteAccessor.java | 509 +++++++++++++++++++
.../pxf/plugins/ignite/IgniteFilterBuilder.java | 147 ++++++
.../ignite/IgnitePartitionFragmenter.java | 402 +++++++++++++++
.../hawq/pxf/plugins/ignite/IgnitePlugin.java | 89 ++++
.../hawq/pxf/plugins/ignite/IgniteResolver.java | 223 ++++++++
.../pxf/plugins/ignite/WhereSQLBuilder.java | 168 ++++++
.../pxf/plugins/ignite/IgniteAccessorTest.java | 130 +++++
.../plugins/ignite/IgniteFilterBuilderTest.java | 100 ++++
.../ignite/IgnitePartitionFragmenterTest.java | 259 ++++++++++
.../hawq/pxf/plugins/ignite/SqlBuilderTest.java | 162 ++++++
.../pxf-private-cdh.classpath.template | 5 +-
.../pxf-private-hdp.classpath.template | 5 +-
.../pxf-private-tar.classpath.template | 3 +
.../src/main/resources/pxf-private.classpath | 3 +
.../main/resources/pxf-privatebigtop.classpath | 5 +-
.../src/main/resources/pxf-privatehdp.classpath | 5 +-
.../src/main/resources/pxf-privatephd.classpath | 5 +-
.../src/main/resources/pxf-profiles-default.xml | 9 +
.../src/main/resources/pxf-profiles.xml | 2 +-
pxf/settings.gradle | 1 +
22 files changed, 2336 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/build.gradle
----------------------------------------------------------------------
diff --git a/pxf/build.gradle b/pxf/build.gradle
index 1ed42bf..3205967 100644
--- a/pxf/build.gradle
+++ b/pxf/build.gradle
@@ -334,6 +334,7 @@ project('pxf') {
requires(versionedPackageName('pxf-hbase'), project.version, GREATER | EQUAL)
requires(versionedPackageName('pxf-json'), project.version, GREATER | EQUAL)
requires(versionedPackageName('pxf-jdbc'), project.version, GREATER | EQUAL)
+ requires(versionedPackageName('pxf-ignite'), project.version, GREATER | EQUAL)
preInstall file('src/scripts/pre-install.sh')
postInstall file('src/scripts/post-install.sh')
@@ -478,7 +479,6 @@ project('pxf-hbase') {
}
}
-
project('pxf-jdbc') {
dependencies {
compile(project(':pxf-api'))
@@ -510,6 +510,36 @@ project('pxf-jdbc') {
}
}
+project('pxf-ignite') {
+ dependencies {
+ compile(project(':pxf-api'))
+ compile(project(':pxf-service'))
+ compile 'com.google.code.gson:gson:2.8.2'
+ compile 'org.apache.commons:commons-compress:1.16.1'
+ testCompile 'com.google.code.gson:gson:2.8.2'
+ testCompile 'org.apache.commons:commons-compress:1.16.1'
+ }
+
+ ospackage {
+ packageName = versionedPackageName("${project.name}")
+ summary = 'HAWQ Extension Framework (PXF), Apache Ignite plugin'
+ description = 'Querying external data stored in the Apache Ignite database.'
+ packager = ' '
+ packageGroup = 'Development/Libraries'
+ release = buildNumber() + '.' + project.osFamily
+ buildHost = ' '
+
+ requires(versionedPackageName('pxf-service'), project.version, GREATER | EQUAL)
+
+ from(jar.outputs.files) {
+ into "/usr/lib/pxf-${project.version}"
+ }
+
+ link("/usr/lib/pxf-${project.version}/${project.name}.jar", "${project.name}-${project.version}.jar")
+ }
+}
+
+
def buildNumber() {
System.getenv('BUILD_NUMBER') ?: System.getProperty('user.name')
}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/README.md
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/README.md b/pxf/pxf-ignite/README.md
new file mode 100644
index 0000000..0effc52
--- /dev/null
+++ b/pxf/pxf-ignite/README.md
@@ -0,0 +1,79 @@
+# Accessing Ignite database using PXF
+
+The PXF Ignite plug-in enables to access the [Apache Ignite database](https://ignite.apache.org/) (both read and write operations are supported) via REST API.
+
+
+## Prerequisites
+
+Check the following before using the plug-in:
+
+* The Ignite plug-in is installed on all PXF nodes;
+
+* The Apache Ignite client is installed and running at the `IGNITE_HOST` (`localhost` by default; this can be changed, see syntax below), and it accepts http queries from the PXF (note that *enabling Ignite REST API does not require changes in Ignite configuration*; see the instruction on how to do that at https://apacheignite.readme.io/docs/rest-api#section-getting-started).
+
+
+## Syntax
+
+```
+CREATE [READABLE | WRITABLE] EXTERNAL TABLE <table_name> (
+ <column_name> <data_type>[, <column_name> <data_type>, ...] | LIKE <other_table>
+)
+LOCATION ('pxf://<ignite_table_name>?PROFILE=Ignite[&<extra-parameter>&<extra-parameter>&...]')
+FORMAT 'CUSTOM' (formatter='pxfwritable_import');
+```
+where each `<extra-parameter>` is one of the following:
+* `IGNITE_HOST=<ignite_host_address_with_port>`. The location of Ignite client node. If not given, `127.0.0.1:8080` is used by default;
+* `IGNITE_CACHE=<ignite_cache_name>`. The name of Ignite cache to use. If not given, this parameter is not included in queries from PXF to Ignite, thus Ignite default values will be used (at the moment, this is `Default` cache). This option is **case-sensitive**;
+* `BUFFER_SIZE=<unsigned_int>`. The number of tuples send to (from) Ignite per a response. The same number of tuples is stored in in-plug-in cache. The values `0` and `1` are equal (cache is not used, each tuple is passed in its own query to Ignite). If not given, `128` is used by default;
+* `PARTITION_BY=<column>:<column_type>`. See below;
+* `RANGE=<start_value>:<end_value>`. See below;
+* `INTERVAL=<value>[:<unit>]`. See below.
+
+
+## Write access
+
+PXF Ignite plugin supports `INSERT` queries. However, due to the usage of REST API, this function has a technical limit: URL can not be longer than approx. 2000 characters. This makes the `INSERT` of very long tuples of data impossible.
+
+Due to this limitation, the recommended value of `BUFFER_SIZE` for `INSERT` queries is `1`. Note that this slightly decreases the perfomance.
+
+
+## Partitioning
+### Introduction
+
+PXF Ignite plugin supports simultaneous **read** access to the Apache Ignite database from multiple PXF segments. *Partitioning* should be used in order to perform such operation.
+
+This feature is optional. If partitioning is not used, all the data will be retrieved by a single PXF segment.
+
+
+### Mechanism
+
+Partitioning in PXF Ignite plug-in works just like in PXF JDBC plug-in.
+
+If partitioning is activated (a valid set of the required parameters is present in the `EXTERNAL TABLE` description; see syntax below), the SELECT query is split into a set of small queries, each of which is called a *fragment*. All the fragments are processed by separate PXF instances simultaneously. If there are more fragments than PXF instances, some instances will process more than one fragment; if only one PXF instance is available, it will process all the fragments.
+
+Extra constraints (`WHERE` expressions) are automatically added to each fragment to guarantee that every tuple of data is retrieved from the Apache Ignite database exactly once.
+
+
+### Syntax
+
+To use partitions, add a set of `<ignite-parameter>`s:
+```
+&PARTITION_BY=<column>:<column_type>&RANGE=<start_value>:<end_value>[&INTERVAL=<value>[:<unit>]]
+```
+
+* The `PARTITION_BY` parameter indicates which column to use as the partition column. Only one column can be used as a partition column.
+ * The `<column>` is the name of a partition column;
+ * The `<column_type>` is the datatype of a partition column. At the moment, the **supported types** are `INT`, `DATE` and `ENUM`. The `DATE` format is `yyyy-MM-dd`.
+
+* The `RANGE` parameter indicates the range of data to be queried. If the partition type is `ENUM`, the `RANGE` parameter must be a list of values, each of which forms its own fragment. In case of `INT` and `DATE` partitions, this parameter must be a finite left-closed range ("infinity" values are not supported):
+ * `[ <start_value> ; <end_value> )`
+ * `... >= start_value AND ... < end_value`;
+
+* The `INTERVAL` parameter is **required** for `INT` and `DATE` partitions. This parameter is ignored if `<column_type>` is `ENUM`.
+ * The `<value>` is the size of each fragment (the last one may be smaller). Note that by default PXF does not support more than 100 fragments;
+ * The `<unit>` **must** be provided if `<column_type>` is `DATE`. At the moment, only `year`, `month` and `day` are supported. This parameter is ignored in case of any other `<column_type>`.
+
+Example partitions:
+* `&PARTITION_BY=id:int&RANGE=42:142&INTERVAL=2`
+* `&PARTITION_BY=createdate:date&RANGE=2008-01-01:2010-01-01&INTERVAL=1:month`
+* `&PARTITION_BY=grade:enum&RANGE=excellent:good:general:bad`
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteAccessor.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteAccessor.java b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteAccessor.java
new file mode 100644
index 0000000..249245b
--- /dev/null
+++ b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteAccessor.java
@@ -0,0 +1,509 @@
+package org.apache.hawq.pxf.plugins.ignite;
+
+/*
+ * 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.
+ */
+
+import org.apache.hawq.pxf.api.OneRow;
+import org.apache.hawq.pxf.api.ReadAccessor;
+import org.apache.hawq.pxf.api.WriteAccessor;
+import org.apache.hawq.pxf.api.UserDataException;
+import org.apache.hawq.pxf.api.utilities.ColumnDescriptor;
+import org.apache.hawq.pxf.api.utilities.InputData;
+import org.apache.hawq.pxf.plugins.ignite.IgnitePlugin;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.io.InputStreamReader;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.net.URL;
+import java.net.URLEncoder;
+import java.net.MalformedURLException;
+import java.net.ProtocolException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import com.google.gson.JsonParser;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonArray;
+
+/**
+ * PXF-Ignite accessor class
+ */
+public class IgniteAccessor extends IgnitePlugin implements ReadAccessor, WriteAccessor {
+ /**
+ * Class constructor
+ */
+ public IgniteAccessor(InputData inputData) throws UserDataException {
+ super(inputData);
+ }
+
+ /**
+ * openForRead() implementation
+ */
+ @Override
+ public boolean openForRead() throws Exception {
+ if (bufferSize == 0) {
+ bufferSize = 1;
+ }
+
+ StringBuilder sb = new StringBuilder();
+
+ // Insert a list of fields to be selected
+ ArrayList<ColumnDescriptor> columns = inputData.getTupleDescription();
+ if (columns == null) {
+ throw new UserDataException("Tuple description must be present.");
+ }
+ sb.append("SELECT ");
+ for (int i = 0; i < columns.size(); i++) {
+ ColumnDescriptor column = columns.get(i);
+ if (i > 0) {
+ sb.append(", ");
+ }
+ sb.append(column.columnName());
+ }
+
+ // Insert the name of the table to select values from
+ sb.append(" FROM ");
+ String tableName = inputData.getDataSource();
+ if (tableName == null) {
+ throw new UserDataException("Table name must be set as DataSource.");
+ }
+ sb.append(tableName);
+
+ // Insert query constraints
+ // Note: Filter constants may be passed to Ignite separately from the WHERE expression, primarily for the safety of the SQL queries. However, at the moment they are passed in the query.
+ ArrayList<String> filterConstants = null;
+ if (inputData.hasFilter()) {
+ WhereSQLBuilder filterBuilder = new WhereSQLBuilder(inputData);
+ String whereSql = filterBuilder.buildWhereSQL();
+
+ if (whereSql != null) {
+ sb.append(" WHERE ").append(whereSql);
+ }
+ }
+
+ // Insert partition constraints
+ IgnitePartitionFragmenter.buildFragmenterSql(inputData, sb);
+
+ // Format URL
+ urlReadStart = buildQueryFldexe(sb.toString(), filterConstants);
+
+ // Send the first REST request that opens the connection
+ JsonElement response = sendRestRequest(urlReadStart);
+
+ // Build 'urlReadFetch' and 'urlReadClose'
+ isLastReadFinished = response.getAsJsonObject().get("last").getAsBoolean();
+ urlReadFetch = buildQueryFetch(response.getAsJsonObject().get("queryId").getAsInt());
+ urlReadClose = buildQueryCls(response.getAsJsonObject().get("queryId").getAsInt());
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ignite read request. URL: '" + urlReadStart + "'");
+ }
+ return true;
+ }
+
+ /**
+ * readNextObject() implementation
+ */
+ @Override
+ public OneRow readNextObject() throws Exception {
+ if (urlReadFetch == null) {
+ LOG.error("readNextObject(): urlReadFetch is null. This means the Ignite qryfldexe query was not executed properly");
+ throw new ProtocolException("readNextObject(): urlReadFetch is null. This means the Ignite qryfldexe query was not executed properly");
+ }
+
+ if (bufferRead.isEmpty()) {
+ // Refill buffer
+ if (isLastReadFinished) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("readNextObject(): All the data received from Ignite");
+ }
+ return null;
+ }
+
+ JsonElement response = sendRestRequest(urlReadFetch);
+ isLastReadFinished = response.getAsJsonObject().get("last").getAsBoolean();
+
+ // Parse 'items'
+ Iterator<JsonElement> itemsIterator = response.getAsJsonObject().get("items").getAsJsonArray().iterator();
+ while (itemsIterator.hasNext()) {
+ if (!bufferRead.add(itemsIterator.next().getAsJsonArray())) {
+ throw new IOException("readNextObject(): not enough memory in 'bufferRead'");
+ }
+ }
+
+ // Check again in case "response" contains no elements
+ if (bufferRead.isEmpty()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("readNextObject(): Buffer refill failed");
+ LOG.debug("readNextObject(): All the data received from Ignite");
+ }
+ return null;
+ }
+ }
+
+ return new OneRow(bufferRead.pollFirst());
+ }
+
+ /**
+ * closeForRead() implementation
+ */
+ @Override
+ public void closeForRead() {
+ if (urlReadClose != null) {
+ try {
+ sendRestRequest(urlReadClose);
+ }
+ catch (Exception e) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("closeForRead() Exception: " + e.getClass().getSimpleName());
+ }
+ }
+ }
+ isLastReadFinished = false;
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ignite read request finished. URL: '" + urlReadClose + "'");
+ }
+ }
+
+ /**
+ * openForWrite() implementation.
+ * No queries are sent to Ignite by this procedure, so if there are some problems (for example, with connection), they will be revealed only during the execution of 'writeNextObject()'
+ */
+ @Override
+ public boolean openForWrite() throws UserDataException {
+ // This is a temporary solution. At the moment there is no other way (except for the usage of user-defined parameters) to get the correct name of Ignite table: GPDB inserts extra data into the address, as required by Hadoop.
+ // Note that if no extra data is present, the 'definedSource' will be left unchanged
+ String definedSource = inputData.getDataSource();
+ Matcher matcher = writeAddressPattern.matcher(definedSource);
+ if (matcher.find()) {
+ inputData.setDataSource(matcher.group(1));
+ }
+
+ StringBuilder sb = new StringBuilder();
+ sb.append("INSERT INTO ");
+
+ // Insert the table name
+ String tableName = inputData.getDataSource();
+ if (tableName == null) {
+ throw new UserDataException("Table name must be set as DataSource.");
+ }
+ sb.append(tableName);
+
+ // Insert the column names
+ sb.append("(");
+ ArrayList<ColumnDescriptor> columns = inputData.getTupleDescription();
+ if (columns == null) {
+ throw new UserDataException("Tuple description must be present.");
+ }
+ String fieldDivisor = "";
+ for (int i = 0; i < columns.size(); i++) {
+ sb.append(fieldDivisor);
+ fieldDivisor = ", ";
+ sb.append(columns.get(i).columnName());
+ }
+ sb.append(")");
+
+ sb.append(" VALUES ");
+
+ queryWrite = sb.toString();
+ return true;
+ }
+
+ /**
+ * writeNextObject() implementation
+ */
+ @Override
+ public boolean writeNextObject(OneRow currentRow) throws Exception {
+ boolean currentRowInBuffer = bufferWrite.add(currentRow);
+
+ if (!isWriteActive) {
+ if (!currentRowInBuffer) {
+ LOG.error("writeNextObject(): Failed (not enough memory in 'bufferWrite')");
+ throw new IOException("writeNextObject(): not enough memory in 'bufferWrite'");
+ }
+ LOG.info("Ignite write request. Query: '" + queryWrite + "'");
+ sendInsertRestRequest(queryWrite);
+ isWriteActive = true;
+ return true;
+ }
+
+ if ((bufferWrite.size() >= bufferSize) || (!currentRowInBuffer)) {
+ sendInsertRestRequest(queryWrite);
+ }
+
+ if (!currentRowInBuffer) {
+ if (!bufferWrite.add(currentRow)) {
+ LOG.error("writeNextObject(): Failed (not enough memory in 'bufferSend')");
+ throw new IOException("writeNextObject(): not enough memory in 'bufferSend'");
+ }
+ }
+
+ return true;
+ }
+
+ /**
+ * closeForWrite() implementation
+ */
+ @Override
+ public void closeForWrite() throws Exception {
+ isWriteActive = false;
+ if (!bufferWrite.isEmpty()) {
+ sendInsertRestRequest(queryWrite);
+ }
+ if (isWriteActive) {
+ // At this point, the request must have finished successfully
+ LOG.info("Ignite write request finished successfully. Query: '" + queryWrite + "'");
+ }
+ }
+
+
+ private static final Log LOG = LogFactory.getLog(IgniteAccessor.class);
+
+ // A pattern to cut extra parameters from 'InputData.dataSource' when write operation is performed. See {@link openForWrite()} for the details
+ private static final Pattern writeAddressPattern = Pattern.compile("/(.*)/[0-9]*-[0-9]*_[0-9]*");
+
+ // Prepared URLs to send to Ignite when reading data
+ private String urlReadStart = null;
+ private String urlReadFetch = null;
+ private String urlReadClose = null;
+ // Set to true when Ignite reported all the data for the SELECT query was retreived
+ private boolean isLastReadFinished = false;
+ // A buffer to store the SELECT query results (without Ignite metadata)
+ private LinkedList<JsonArray> bufferRead = new LinkedList<JsonArray>();
+
+ // A template for the INSERT
+ private String queryWrite = null;
+ // Set to true when the INSERT operation is in progress
+ private boolean isWriteActive = false;
+ // A buffer to store prepared values for the INSERT query
+ private LinkedList<OneRow> bufferWrite = new LinkedList<OneRow>();
+
+ /**
+ * Build HTTP GET query for Ignite REST API with command 'qryfldexe'
+ *
+ * @param querySql SQL query
+ * @param filterConstants A list of Constraints' constants. Must be null in this version.
+ *
+ * @return Prepared HTTP query. The query will be properly encoded with {@link java.net.URLEncoder}
+ *
+ * @throws UnsupportedEncodingException from {@link java.net.URLEncoder.encode()}
+ */
+ private String buildQueryFldexe(String querySql, List<String> filterConstants) throws UnsupportedEncodingException {
+ StringBuilder sb = new StringBuilder();
+ sb.append("http://");
+ sb.append(igniteHost);
+ sb.append("/ignite");
+ sb.append("?");
+ sb.append("cmd=qryfldexe");
+ sb.append("&");
+ sb.append("pageSize=0");
+ sb.append("&");
+ if (cacheName != null) {
+ sb.append("cacheName=");
+ // Note that Ignite supports only "good" cache names (those that will be left unchanged by the URLEncoder.encode())
+ sb.append(URLEncoder.encode(cacheName, "UTF-8"));
+ sb.append("&");
+ }
+ /*
+ 'filterConstants' must always be null in the current version.
+ This code allows to pass filters' constants separately from the filters' expressions. This feature is supported by Ignite database; however, it is not implemented in PXF Ignite plugin at the moment.
+ To implement this, changes should be made in {@link WhereSQLBuilder} (form SQL query without filter constants) and {@link IgnitePartitionFragmenter} (form partition constraints the similar way).
+ */
+ int counter = 1;
+ if (filterConstants != null) {
+ for (String constant : filterConstants) {
+ sb.append("arg");
+ sb.append(counter);
+ sb.append("=");
+ sb.append(URLEncoder.encode(constant, "UTF-8"));
+ sb.append("&");
+ counter += 1;
+ }
+ }
+ sb.append("qry=");
+ sb.append(URLEncoder.encode(querySql, "UTF-8"));
+
+ return sb.toString();
+ }
+
+ /**
+ * Build HTTP GET query for Ignite REST API with command 'qryfetch'
+ * This query is used to retrieve data after the 'qryfldexe' command started
+ *
+ * @param queryId ID of the query assigned by Ignite when the query started
+ *
+ * @return Prepared HTTP query
+ */
+ private String buildQueryFetch(int queryId) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("http://");
+ sb.append(igniteHost);
+ sb.append("/ignite");
+ sb.append("?");
+ sb.append("cmd=qryfetch");
+ sb.append("&");
+ sb.append("pageSize=");
+ sb.append(bufferSize);
+ sb.append("&");
+ sb.append("qryId=");
+ sb.append(queryId);
+
+ return sb.toString();
+ }
+
+ /**
+ * Build HTTP GET query for Ignite REST API with command 'qrycls'
+ * This query is used to close query resources on Ignite side
+ *
+ * @param queryId ID of the query assigned by Ignite when the query started
+ *
+ * @return Prepared HTTP query
+ */
+ private String buildQueryCls(int queryId) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("http://");
+ sb.append(igniteHost);
+ sb.append("/ignite");
+ sb.append("?");
+ sb.append("cmd=qrycls");
+ sb.append("&");
+ sb.append("qryId=");
+ sb.append(queryId);
+
+ return sb.toString();
+ }
+
+ /**
+ * Send a REST request to the Ignite server
+ *
+ * @param query A prepared and properly encoded HTTP GET request
+ *
+ * @return "response" field from the received JSON object
+ * (See Ignite REST API documentation for details)
+ *
+ * @throws ProtocolException if Ignite reports error in it's JSON response
+ * @throws MalformedURLException if URL is malformed
+ * @throws IOException in case of connection failure
+ */
+ private JsonElement sendRestRequest(String query) throws ProtocolException, MalformedURLException, IOException {
+ // Create URL object. This operation may throw 'MalformedURLException'
+ URL url = new URL(query);
+
+ // Connect to the Ignite server, send query and get raw response
+ BufferedReader reader = null;
+ String responseRaw = null;
+ try {
+ StringBuilder sb = new StringBuilder();
+ reader = new BufferedReader(new InputStreamReader(url.openStream()));
+ String responseLine;
+ while ((responseLine = reader.readLine()) != null) {
+ sb.append(responseLine);
+ }
+ responseRaw = sb.toString();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("sendRestRequest(): URL: '" + query + "'; Result: '" + responseRaw + "'");
+ }
+ }
+ catch (Exception e) {
+ LOG.error("sendRestRequest(): Failed (connection failure). URL is '" + query + "'");
+ throw e;
+ }
+ finally {
+ if (reader != null) {
+ reader.close();
+ }
+ // if 'reader' is null, an exception must have been thrown
+ }
+
+ // Parse raw Ignite server response
+ JsonElement response = null;
+ String error = null;
+ int successStatus;
+ try {
+ response = new JsonParser().parse(responseRaw);
+
+ if (!response.getAsJsonObject().get("error").isJsonNull()) {
+ error = response.getAsJsonObject().get("error").getAsString();
+ }
+ successStatus = response.getAsJsonObject().get("successStatus").getAsInt();
+ }
+ catch (Exception e) {
+ LOG.error("sendRestRequest(): Failed (JSON parsing failure). URL is '" + query + "'");
+ throw e;
+ }
+
+ // Check errors reported by Ignite
+ if ((error != null) || (successStatus != 0)) {
+ LOG.error("sendRestRequest(): Failed (failure on Ignite side: '" + error + "'). URL is '" + query + "'");
+ throw new ProtocolException("Ignite failure: status " + successStatus + ", '" + error + "'");
+ }
+
+ // Return response without metadata
+ try {
+ return response.getAsJsonObject().get("response");
+ }
+ catch (Exception e) {
+ LOG.error("sendRestRequest(): Failed (JSON parsing failure). URL is '" + query + "'");
+ throw e;
+ }
+ }
+
+ /**
+ * Send an INSERT REST request to the Ignite server.
+ *
+ * Note that
+ *
+ * The {@link sendRestRequest()} is used to handle network operations, thus all its exceptions may be thrown. They are:
+ * @throws ProtocolException if Ignite reports error in it's JSON response
+ * @throws MalformedURLException if URL is malformed
+ * @throws IOException in case of connection failure
+ */
+ private void sendInsertRestRequest(String query) throws ProtocolException, MalformedURLException, IOException {
+ if (query == null) {
+ LOG.error("sendInsertRestRequest(): Failed (malformed URL). URL is null");
+ throw new MalformedURLException("sendInsertRestRequest(): query is null");
+ }
+ if (bufferWrite.isEmpty()) {
+ return;
+ }
+
+ StringBuilder sb = new StringBuilder(query);
+ String fieldDivisor = "";
+ for (OneRow row : bufferWrite) {
+ sb.append(fieldDivisor);
+ fieldDivisor = ", ";
+ sb.append((String)row.getData());
+ }
+ bufferWrite.clear();
+
+ // Send REST request 'qryfldexe' to Ignite
+ JsonElement response = sendRestRequest(buildQueryFldexe(sb.toString(), null));
+ // Close the request immediately
+ sendRestRequest(buildQueryCls(response.getAsJsonObject().get("queryId").getAsInt()));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteFilterBuilder.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteFilterBuilder.java b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteFilterBuilder.java
new file mode 100644
index 0000000..98b36b6
--- /dev/null
+++ b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteFilterBuilder.java
@@ -0,0 +1,147 @@
+package org.apache.hawq.pxf.plugins.ignite;
+
+/*
+ * 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.
+ */
+
+import org.apache.hawq.pxf.api.BasicFilter;
+import org.apache.hawq.pxf.api.FilterParser;
+import org.apache.hawq.pxf.api.LogicalFilter;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Uses the filter parser code to build a filter object, either simple - a
+ * single {@link BasicFilter} object or a
+ * compound - a {@link List} of
+ * {@link BasicFilter} objects.
+ * The subclass {@link WhereSQLBuilder} will use the filter for
+ * generate WHERE statement.
+ */
+public class IgniteFilterBuilder implements FilterParser.FilterBuilder {
+ /**
+ * Translates a filterString into a {@link BasicFilter} or a
+ * list of such filters.
+ *
+ * @param filterString the string representation of the filter
+ * @return a single {@link BasicFilter}
+ * object or a {@link List} of
+ * {@link BasicFilter} objects.
+ * @throws Exception if parsing the filter failed or filter is not a basic
+ * filter or list of basic filters
+ */
+ public Object getFilterObject(String filterString) throws Exception {
+ FilterParser parser = new FilterParser(this);
+ Object result = parser.parse(filterString.getBytes(FilterParser.DEFAULT_CHARSET));
+
+ if (!(result instanceof LogicalFilter) && !(result instanceof BasicFilter)
+ && !(result instanceof List)) {
+ throw new Exception("String " + filterString
+ + " resolved to no filter");
+ }
+
+ return result;
+ }
+
+
+ @Override
+ public Object build(FilterParser.LogicalOperation op, Object leftOperand, Object rightOperand) {
+ return handleLogicalOperation(op, leftOperand, rightOperand);
+ }
+
+ @Override
+ public Object build(FilterParser.LogicalOperation op, Object filter) {
+ return handleLogicalOperation(op, filter);
+ }
+
+ @Override
+ public Object build(FilterParser.Operation opId, Object leftOperand,
+ Object rightOperand) throws Exception {
+ // Assume column is on the left
+ return handleSimpleOperations(opId,
+ (FilterParser.ColumnIndex) leftOperand,
+ (FilterParser.Constant) rightOperand);
+ }
+
+ @Override
+ public Object build(FilterParser.Operation operation, Object operand) throws Exception {
+ if (operation == FilterParser.Operation.HDOP_IS_NULL || operation == FilterParser.Operation.HDOP_IS_NOT_NULL) {
+ // use null for the constant value of null comparison
+ return handleSimpleOperations(operation, (FilterParser.ColumnIndex) operand, null);
+ } else {
+ throw new Exception("Unsupported unary operation " + operation);
+ }
+ }
+
+ /*
+ * Handles simple column-operator-constant expressions Creates a special
+ * filter in the case the column is the row key column
+ */
+ private BasicFilter handleSimpleOperations(FilterParser.Operation opId,
+ FilterParser.ColumnIndex column,
+ FilterParser.Constant constant) {
+ return new BasicFilter(opId, column, constant);
+ }
+
+ /**
+ * Handles AND of already calculated expressions. Currently only AND can be handled
+ *
+ * Four cases here:
+ * <ol>
+ * <li>both are simple filters</li>
+ * <li>left is a FilterList and right is a filter</li>
+ * <li>left is a filter and right is a FilterList</li>
+ * <li>both are FilterLists</li>
+ * </ol>
+ * Currently, 1, 2 can occur, since no parenthesis are used
+ *
+ * @param left left hand filter
+ * @param right right hand filter
+ * @return list of filters constructing the filter tree
+ */
+ private List<BasicFilter> handleCompoundOperations(List<BasicFilter> left,
+ BasicFilter right) {
+ left.add(right);
+ return left;
+ }
+
+ private List<BasicFilter> handleCompoundOperations(BasicFilter left,
+ BasicFilter right) {
+ List<BasicFilter> result = new LinkedList<>();
+
+ result.add(left);
+ result.add(right);
+
+ return result;
+ }
+
+ private Object handleLogicalOperation(FilterParser.LogicalOperation operator, Object leftOperand, Object rightOperand) {
+
+ List<Object> result = new LinkedList<>();
+
+ result.add(leftOperand);
+ result.add(rightOperand);
+ return new LogicalFilter(operator, result);
+ }
+
+ private Object handleLogicalOperation(FilterParser.LogicalOperation operator, Object filter) {
+ return new LogicalFilter(operator, Arrays.asList(filter));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgnitePartitionFragmenter.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgnitePartitionFragmenter.java b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgnitePartitionFragmenter.java
new file mode 100644
index 0000000..0fb0b33
--- /dev/null
+++ b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgnitePartitionFragmenter.java
@@ -0,0 +1,402 @@
+package org.apache.hawq.pxf.plugins.ignite;
+
+/*
+ * 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.
+ */
+
+import org.apache.hawq.pxf.api.Fragment;
+import org.apache.hawq.pxf.api.Fragmenter;
+import org.apache.hawq.pxf.api.FragmentsStats;
+import org.apache.hawq.pxf.api.UserDataException;
+import org.apache.hawq.pxf.api.utilities.InputData;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.commons.compress.utils.ByteUtils;
+
+/**
+ * PXF-Ignite fragmenter class
+ *
+ * This fragmenter works just like the one in PXF JDBC plugin
+ */
+public class IgnitePartitionFragmenter extends Fragmenter {
+ /**
+ * Insert partition constraints into the prepared SQL query.
+ *
+ * @param inputData pre-validated PXF InputData
+ * @param sb the SQL query that is prepared for appending extra WHERE constraints.
+ */
+ public static void buildFragmenterSql(InputData inputData, StringBuilder sb) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("buildFragmenterSql() called");
+ }
+
+ if (inputData.getUserProperty("PARTITION_BY") == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("buildFragmenterSql(): Partition is not used");
+ }
+ return;
+ }
+
+ byte[] meta = inputData.getFragmentMetadata();
+ if (meta == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("buildFragmenterSql(): Fragment metadata is null, no partition constraints added");
+ }
+ return;
+ }
+
+ // Note that these parameters have already been validated when constructing fragment.
+ String[] partitionBy = inputData.getUserProperty("PARTITION_BY").split(":");
+ String partitionColumn = partitionBy[0];
+ PartitionType partitionType = PartitionType.typeOf(partitionBy[1]);
+
+ if (!sb.toString().contains("WHERE")) {
+ sb.append(" WHERE ");
+ }
+ else {
+ sb.append(" AND ");
+ }
+
+ switch (partitionType) {
+ case DATE: {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("buildFragmenterSql(): DATE partition found");
+ }
+
+ // Get fragment metadata
+ Date fragStart = new Date(ByteUtils.fromLittleEndian(meta, 0, 8));
+ Date fragEnd = new Date(ByteUtils.fromLittleEndian(meta, 8, 8));
+
+ // Add constraints
+ SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");
+ sb.append(partitionColumn).append(">=").append("'" + df.format(fragStart) + "'");
+ sb.append(" AND ");
+ sb.append(partitionColumn).append("<").append("'" + df.format(fragEnd) + "'");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("buildFragmenterSql(): DATE partition constraints added");
+ }
+ break;
+ }
+ case INT: {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("buildFragmenterSql(): INT partition found");
+ }
+
+ // Get fragment metadata
+ int fragStart = (int)ByteUtils.fromLittleEndian(meta, 0, 4);
+ int fragEnd = (int)ByteUtils.fromLittleEndian(meta, 4, 4);
+
+ // Add constraints
+ sb.append(partitionColumn).append(">=").append(fragStart);
+ sb.append(" AND ");
+ sb.append(partitionColumn).append("<").append(fragEnd);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("buildFragmenterSql(): INT partition constraints added");
+ }
+ break;
+ }
+ case ENUM: {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("buildFragmenterSql(): ENUM partition found");
+ }
+
+ // Add constraints. Fragment metadata should not be parsed.
+ sb.append(partitionColumn).append("='").append(new String(meta)).append("'");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("buildFragmenterSql(): ENUM partition constraints added");
+ }
+ break;
+ }
+ }
+ }
+
+ /**
+ * Class constructor
+ *
+ * @throws UserDataException if the request parameter is malformed
+ */
+ public IgnitePartitionFragmenter(InputData inputData) throws UserDataException {
+ super(inputData);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Constructor started");
+ }
+
+ if (inputData.getUserProperty("PARTITION_BY") == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Constructor successful; partition was not used");
+ }
+ return;
+ }
+
+ try {
+ partitionBy = inputData.getUserProperty("PARTITION_BY").split(":");
+ partitionType = PartitionType.typeOf(partitionBy[1]);
+ }
+ catch (IllegalArgumentException | ArrayIndexOutOfBoundsException e1) {
+ throw new UserDataException("The parameter 'PARTITION_BY' is invalid. The pattern is 'column_name:DATE|INT|ENUM'");
+ }
+
+ // Parse and validate parameter-RANGE
+ try {
+ String rangeStr = inputData.getUserProperty("RANGE");
+ if (rangeStr != null) {
+ range = rangeStr.split(":");
+ if (range.length == 1 && partitionType != PartitionType.ENUM) {
+ throw new UserDataException("The parameter 'RANGE' does not specify '[:end_value]'");
+ }
+ }
+ else {
+ throw new UserDataException("The parameter 'RANGE' must be specified along with 'PARTITION_BY'");
+ }
+ }
+ catch (IllegalArgumentException e) {
+ throw new UserDataException("The parameter 'RANGE' is invalid, the pattern is 'start_value[:end_value]'");
+ }
+
+ // Parse and validate parameter-INTERVAL
+ try {
+ String intervalStr = inputData.getUserProperty("INTERVAL");
+ if (intervalStr != null) {
+ interval = intervalStr.split(":");
+ intervalNum = Integer.parseInt(interval[0]);
+ if (interval.length > 1) {
+ intervalType = IntervalType.typeOf(interval[1]);
+ }
+ if (interval.length == 1 && partitionType == PartitionType.DATE) {
+ throw new UserDataException("The parameter 'INTERVAL' does not specify unit [:year|month|day]");
+ }
+ }
+ else if (partitionType != PartitionType.ENUM) {
+ throw new UserDataException("The parameter 'INTERVAL' must be specified along with 'PARTITION_BY'");
+ }
+ if (intervalNum < 1) {
+ throw new UserDataException("The parameter 'INTERVAL' must be at least 1. The actual is '" + intervalNum + "'");
+ }
+ }
+ catch (IllegalArgumentException e) {
+ throw new UserDataException("The parameter 'INTERVAL' invalid. The pattern is 'interval_num[:interval_unit]'");
+ }
+
+ // Parse date partition
+ try {
+ if (partitionType == PartitionType.DATE) {
+ SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");
+ rangeStart = Calendar.getInstance();
+ rangeStart.setTime(df.parse(range[0]));
+ rangeEnd = Calendar.getInstance();
+ rangeEnd.setTime(df.parse(range[1]));
+ }
+ } catch (ParseException e) {
+ throw new UserDataException("The parameter 'RANGE' has invalid date format. Expected format is 'yyyy-MM-dd'");
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Constructor successful; some partition used");
+ }
+ }
+
+ /**
+ * Returns statistics for the Ignite table. This is not implemented in the current version
+ * @throws UnsupportedOperationException
+ */
+ @Override
+ public FragmentsStats getFragmentsStats() throws UnsupportedOperationException {
+ throw new UnsupportedOperationException("ANALYZE for Ignite plugin is not supported");
+ }
+
+ /**
+ * Returns list of fragments for Ignite table queries
+ *
+ * @throws UnsupportedOperationException if a partition of unknown type was found
+ *
+ * @return a list of fragments
+ */
+ @Override
+ public List<Fragment> getFragments() throws UnsupportedOperationException {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getFragments() called; dataSource is '" + inputData.getDataSource() + "'");
+ }
+
+ byte[] fragmentMetadata = null;
+ byte[] fragmentUserdata = null;
+
+ if (partitionType == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getFragments() found no partition");
+ }
+ Fragment fragment = new Fragment(inputData.getDataSource(), replicaHostAddressWrapped, fragmentMetadata, fragmentUserdata);
+ fragments.add(fragment);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getFragments() successful");
+ }
+ return fragments;
+ }
+
+ switch (partitionType) {
+ case DATE: {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getFragments() found DATE partition");
+ }
+ int currInterval = intervalNum;
+
+ Calendar fragStart = rangeStart;
+ while (fragStart.before(rangeEnd)) {
+ Calendar fragEnd = (Calendar) fragStart.clone();
+ switch (intervalType) {
+ case DAY:
+ fragEnd.add(Calendar.DAY_OF_MONTH, currInterval);
+ break;
+ case MONTH:
+ fragEnd.add(Calendar.MONTH, currInterval);
+ break;
+ case YEAR:
+ fragEnd.add(Calendar.YEAR, currInterval);
+ break;
+ }
+ if (fragEnd.after(rangeEnd))
+ fragEnd = (Calendar) rangeEnd.clone();
+
+ fragmentMetadata = new byte[16];
+ ByteUtils.toLittleEndian(fragmentMetadata, fragStart.getTimeInMillis(), 0, 8);
+ ByteUtils.toLittleEndian(fragmentMetadata, fragEnd.getTimeInMillis(), 8, 8);
+ Fragment fragment = new Fragment(inputData.getDataSource(), replicaHostAddressWrapped, fragmentMetadata, fragmentUserdata);
+
+ fragments.add(fragment);
+
+ // Continue the previous fragment
+ fragStart = fragEnd;
+ }
+ break;
+ }
+ case INT: {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getFragments() found INT partition");
+ }
+ int rangeStart = Integer.parseInt(range[0]);
+ int rangeEnd = Integer.parseInt(range[1]);
+ int currInterval = intervalNum;
+
+ int fragStart = rangeStart;
+ while (fragStart < rangeEnd) {
+ int fragEnd = fragStart + currInterval;
+ if (fragEnd > rangeEnd) {
+ fragEnd = rangeEnd;
+ }
+
+ fragmentMetadata = new byte[8];
+ ByteUtils.toLittleEndian(fragmentMetadata, fragStart, 0, 4);
+ ByteUtils.toLittleEndian(fragmentMetadata, fragEnd, 4, 4);
+ Fragment fragment = new Fragment(inputData.getDataSource(), replicaHostAddressWrapped, fragmentMetadata, fragmentUserdata);
+
+ fragments.add(fragment);
+
+ // Continue the previous fragment
+ fragStart = fragEnd;
+ }
+ break;
+ }
+ case ENUM: {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getFragments() found ENUM partition");
+ }
+ for (String frag : range) {
+ fragmentMetadata = frag.getBytes();
+ Fragment fragment = new Fragment(inputData.getDataSource(), replicaHostAddressWrapped, fragmentMetadata, fragmentUserdata);
+ fragments.add(fragment);
+ }
+ break;
+ }
+ default: {
+ throw new UnsupportedOperationException("getFragments() found a partition of unknown type and failed");
+ }
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getFragments() successful");
+ }
+ return fragments;
+ }
+
+
+ /**
+ * Partition type
+ */
+ private static enum PartitionType {
+ DATE,
+ INT,
+ ENUM;
+
+ public static PartitionType typeOf(String str) {
+ return valueOf(str.toUpperCase());
+ }
+ }
+ /**
+ * Partition interval, for 'partitionType == DATE'
+ */
+ private static enum IntervalType {
+ DAY,
+ MONTH,
+ YEAR;
+
+ public static IntervalType typeOf(String str) {
+ return valueOf(str.toUpperCase());
+ }
+ }
+
+ private static final Log LOG = LogFactory.getLog(IgnitePartitionFragmenter.class);
+
+ /**
+ * The replica holder. This is an address of a *PXF* host that processes fragments.
+ * It is always 'localhost'
+ */
+ private static final String[] replicaHostAddressWrapped;
+ static {
+ String[] temp;
+ try {
+ temp = new String[]{InetAddress.getLocalHost().getHostAddress()};
+ }
+ catch (UnknownHostException e) {
+ // In fact, 'localhost' can always be obtained
+ temp = new String[]{"localhost"};
+ }
+ replicaHostAddressWrapped = temp;
+ }
+
+ private String[] partitionBy = null;
+ private String[] range = null;
+ private String[] interval = null;
+ private PartitionType partitionType = null;
+ private IntervalType intervalType = null;
+ private int intervalNum = 1;
+
+ // Used only if 'partitionType' is 'DATE'
+ private Calendar rangeStart = null;
+ private Calendar rangeEnd = null;
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgnitePlugin.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgnitePlugin.java b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgnitePlugin.java
new file mode 100644
index 0000000..8796420
--- /dev/null
+++ b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgnitePlugin.java
@@ -0,0 +1,89 @@
+package org.apache.hawq.pxf.plugins.ignite;
+
+/*
+ * 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.
+ */
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hawq.pxf.api.UserDataException;
+import org.apache.hawq.pxf.api.utilities.InputData;
+import org.apache.hawq.pxf.api.utilities.Plugin;
+
+/**
+ * PXF-Ignite base class.
+ * This class manages the user-defined parameters provided in the query from PXF.
+ * Implemented subclasses: {@link IgniteAccessor}, {@link IgniteResolver}.
+ */
+public class IgnitePlugin extends Plugin {
+ // Ignite cache
+ protected static final String igniteHostDefault = "127.0.0.1:8080";
+ protected String igniteHost = null;
+ // PXF buffer for Ignite data. '0' is allowed for INSERT queries
+ protected static final int bufferSizeDefault = 128;
+ protected int bufferSize = bufferSizeDefault;
+ // Ignite cache name
+ protected String cacheName = null;
+
+ /**
+ * Class constructor. Parses and checks 'InputData'
+ * @param inputData
+ * @throws UserDataException if the request parameter is malformed
+ */
+ public IgnitePlugin(InputData inputData) throws UserDataException {
+ super(inputData);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Constructor started");
+ }
+
+ igniteHost = inputData.getUserProperty("IGNITE_HOST");
+ if (igniteHost == null) {
+ igniteHost = igniteHostDefault;
+ }
+
+ cacheName = inputData.getUserProperty("IGNITE_CACHE");
+ // If this value is null, Ignite will use the default cache
+
+ String bufferSize_str = inputData.getUserProperty("BUFFER_SIZE");
+ if (bufferSize_str != null) {
+ try {
+ bufferSize = Integer.parseInt(bufferSize_str);
+ // Zero value is allowed for INSERT queries
+ if (bufferSize < 0) {
+ bufferSize = bufferSizeDefault;
+ LOG.warn("Buffer size is incorrect; set to the default value (" + bufferSizeDefault + ")");
+ }
+ }
+ catch (NumberFormatException e) {
+ bufferSize = bufferSizeDefault;
+ LOG.warn("Buffer size is incorrect; set to the default value (" + bufferSizeDefault + ")");
+ }
+ }
+ // else: bufferSize is already set to bufferSizeDefault
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Constructor successful");
+ }
+ }
+
+ public boolean isThreadSafe() {
+ return true;
+ }
+
+ private static final Log LOG = LogFactory.getLog(IgnitePlugin.class);
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteResolver.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteResolver.java b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteResolver.java
new file mode 100644
index 0000000..9424fe5
--- /dev/null
+++ b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/IgniteResolver.java
@@ -0,0 +1,223 @@
+package org.apache.hawq.pxf.plugins.ignite;
+
+/*
+ * 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.
+ */
+
+import org.apache.hawq.pxf.api.OneRow;
+import org.apache.hawq.pxf.api.OneField;
+import org.apache.hawq.pxf.api.io.DataType;
+import org.apache.hawq.pxf.api.utilities.ColumnDescriptor;
+import org.apache.hawq.pxf.api.utilities.InputData;
+import org.apache.hawq.pxf.api.ReadResolver;
+import org.apache.hawq.pxf.api.WriteResolver;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.codec.binary.Hex;
+
+import com.google.gson.JsonArray;
+
+/**
+ * PXF-Ignite resolver class
+ */
+public class IgniteResolver extends IgnitePlugin implements ReadResolver, WriteResolver {
+ /**
+ * Class constructor
+ */
+ public IgniteResolver(InputData input) throws Exception {
+ super(input);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Constructor started");
+ }
+
+ columns = input.getTupleDescription();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Constructor successful");
+ }
+ }
+
+ /**
+ * Transform a {@link JsonArray} stored in {@link OneRow} into a list of {@link OneField}
+ *
+ * @throws ParseException if the response could not be correctly parsed
+ * @throws UnsupportedOperationException if the type of some field is not supported
+ */
+ @Override
+ public List<OneField> getFields(OneRow row) throws ParseException, UnsupportedOperationException {
+ JsonArray result = JsonArray.class.cast(row.getData());
+ LinkedList<OneField> fields = new LinkedList<OneField>();
+
+ if (result.size() != columns.size()) {
+ throw new ParseException("getFields(): Failed (a tuple received from Ignite contains more or less fields than requested). Raw tuple: '" + result.toString() + "'", 0);
+ }
+
+ for (int i = 0; i < columns.size(); i++) {
+ Object value = null;
+ OneField oneField = new OneField(columns.get(i).columnTypeCode(), null);
+
+ // Handle null values
+ if (result.get(i).isJsonNull()) {
+ fields.add(oneField);
+ continue;
+ }
+ switch (DataType.get(oneField.type)) {
+ case INTEGER:
+ value = result.get(i).getAsInt();
+ break;
+ case FLOAT8:
+ value = result.get(i).getAsDouble();
+ break;
+ case REAL:
+ value = result.get(i).getAsFloat();
+ break;
+ case BIGINT:
+ value = result.get(i).getAsLong();
+ break;
+ case SMALLINT:
+ value = result.get(i).getAsShort();
+ break;
+ case BOOLEAN:
+ value = result.get(i).getAsBoolean();
+ break;
+ case VARCHAR:
+ case BPCHAR:
+ case TEXT:
+ case NUMERIC:
+ value = result.get(i).getAsString();
+ break;
+ case BYTEA:
+ value = Base64.decodeBase64(result.get(i).getAsString());
+ break;
+ case TIMESTAMP:
+ boolean isConversionSuccessful = false;
+ for (SimpleDateFormat sdf : getTimestampSDFs.get()) {
+ try {
+ value = sdf.parse(result.get(i).getAsString());
+ isConversionSuccessful = true;
+ break;
+ }
+ catch (ParseException e) {
+ // pass
+ }
+ }
+ if (!isConversionSuccessful) {
+ throw new ParseException(result.get(i).getAsString(), 0);
+ }
+ break;
+ case DATE:
+ value = getDateSDF.parse(result.get(i).getAsString());
+ break;
+ default:
+ throw new UnsupportedOperationException("Field type not supported: " + DataType.get(oneField.type).toString()
+ + ", Column: " + columns.get(i).columnName());
+ }
+
+ oneField.val = value;
+ fields.add(oneField);
+ }
+
+ return fields;
+ }
+
+ /**
+ * Transforms a list of {@link OneField} from PXF into a {@link OneRow} with a string inside, containing a tuple from SQL INSERT query
+ *
+ * @throws UnsupportedOperationException if the type of some field is not supported
+ */
+ @Override
+ public OneRow setFields(List<OneField> record) throws UnsupportedOperationException {
+ StringBuilder sb = new StringBuilder();
+ String fieldDivisor = "";
+
+ sb.append("(");
+ for (OneField oneField : record) {
+ sb.append(fieldDivisor);
+ fieldDivisor = ", ";
+ switch (DataType.get(oneField.type)) {
+ case BOOLEAN:
+ case INTEGER:
+ case FLOAT8:
+ case REAL:
+ case BIGINT:
+ case NUMERIC:
+ case SMALLINT:
+ sb.append(String.valueOf(oneField.val));
+ break;
+ case VARCHAR:
+ case BPCHAR:
+ case TEXT:
+ sb.append("'" + String.valueOf(oneField.val) + "'");
+ break;
+ case BYTEA:
+ sb.append("'" + Hex.encodeHexString((byte[])(oneField.val)) + "'");
+ break;
+ case TIMESTAMP:
+ sb.append(setTimestampSDF.get().format(oneField.val));
+ break;
+ case DATE:
+ sb.append(setDateSDF.format(oneField.val));
+ break;
+ default:
+ throw new UnsupportedOperationException("Field type not supported: " + DataType.get(oneField.type).toString());
+ }
+ }
+ sb.append(")");
+ return new OneRow(sb.toString());
+ }
+
+ private static final Log LOG = LogFactory.getLog(IgniteResolver.class);
+
+ private static final SimpleDateFormat getDateSDF = new SimpleDateFormat("yyyy-MM-dd");
+ private static final SimpleDateFormat setDateSDF = new SimpleDateFormat("'yyyy-MM-dd'");
+
+ // SimpleDateFormats to parse Ignite TIMESTAMP format
+ private static ThreadLocal<SimpleDateFormat[]> getTimestampSDFs = new ThreadLocal<SimpleDateFormat[]>() {
+ @Override protected SimpleDateFormat[] initialValue() {
+ SimpleDateFormat[] retRes = {
+ new SimpleDateFormat("MMM d, yyyy hh:mm:ss.SSSSSS a"),
+ new SimpleDateFormat("MMM d, yyyy hh:mm:ss.SSSSS a"),
+ new SimpleDateFormat("MMM d, yyyy hh:mm:ss.SSSS a"),
+ new SimpleDateFormat("MMM d, yyyy hh:mm:ss.SSS a"),
+ new SimpleDateFormat("MMM d, yyyy hh:mm:ss.SS a"),
+ new SimpleDateFormat("MMM d, yyyy hh:mm:ss.S a"),
+ new SimpleDateFormat("MMM d, yyyy hh:mm:ss a")
+ };
+ return retRes;
+ }
+ };
+
+ // SimpleDateFormat to properly encode TIMESTAMP for INSERT queries
+ private static ThreadLocal<SimpleDateFormat> setTimestampSDF = new ThreadLocal<SimpleDateFormat>() {
+ @Override protected SimpleDateFormat initialValue() {
+ return new SimpleDateFormat("'yyyy-MM-dd hh:mm:ss.SSSSSS'");
+ }
+ };
+
+ // HAWQ column descriptors
+ private ArrayList<ColumnDescriptor> columns = null;
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/WhereSQLBuilder.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/WhereSQLBuilder.java b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/WhereSQLBuilder.java
new file mode 100644
index 0000000..01a6980
--- /dev/null
+++ b/pxf/pxf-ignite/src/main/java/org/apache/hawq/pxf/plugins/ignite/WhereSQLBuilder.java
@@ -0,0 +1,168 @@
+package org.apache.hawq.pxf.plugins.ignite;
+/*
+ * 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.
+ */
+
+import org.apache.hawq.pxf.api.LogicalFilter;
+import org.apache.hawq.pxf.api.BasicFilter;
+import org.apache.hawq.pxf.api.FilterParser;
+import org.apache.hawq.pxf.api.io.DataType;
+import org.apache.hawq.pxf.api.utilities.ColumnDescriptor;
+import org.apache.hawq.pxf.api.utilities.InputData;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Parse filter object generated by the parent class {@link IgniteFilterBuilder} and build WHERE statement.
+ * Only HDOP_AND is supported as a logicalFilter at the moment.
+ *
+ * If an error occurs, the filter is not applied at all and no error is returned. Note that in this case the filter will not be applied on Ignite side, but it will be applied by PXF itself (later), thus the results of the query will be correct.
+ */
+public class WhereSQLBuilder extends IgniteFilterBuilder {
+ /**
+ * Class constructor
+ */
+ public WhereSQLBuilder(InputData input) {
+ inputData = input;
+ }
+
+ /**
+ * Build a WHERE statement using the InputData provided to constructor.
+ *
+ * @throws Exception if 'InputData' has filter, but its 'filterString' is incorrect
+ */
+ public String buildWhereSQL() throws Exception {
+ if (!inputData.hasFilter()) {
+ return null;
+ }
+
+ List<BasicFilter> filters = null;
+ try {
+ String filterString = inputData.getFilterString();
+ Object filterObj = getFilterObject(filterString);
+
+ filters = convertBasicFilterList(filterObj, filters);
+
+ StringBuffer sb = new StringBuffer();
+ String andDivisor = "";
+ for (Object obj : filters) {
+ BasicFilter filter = (BasicFilter) obj;
+
+ sb.append(andDivisor);
+ andDivisor = " AND ";
+
+ ColumnDescriptor column = inputData.getColumn(filter.getColumn().index());
+ //the column name of filter
+ sb.append(column.columnName());
+
+ //the operation of filter
+ FilterParser.Operation op = filter.getOperation();
+ switch (op) {
+ case HDOP_LT:
+ sb.append("<");
+ break;
+ case HDOP_GT:
+ sb.append(">");
+ break;
+ case HDOP_LE:
+ sb.append("<=");
+ break;
+ case HDOP_GE:
+ sb.append(">=");
+ break;
+ case HDOP_EQ:
+ sb.append("=");
+ break;
+ default:
+ throw new UnsupportedFilterException("unsupported Filter operation : " + op);
+ }
+
+ Object val = filter.getConstant().constant();
+ switch (DataType.get(column.columnTypeCode())) {
+ case SMALLINT:
+ case INTEGER:
+ case BIGINT:
+ case FLOAT8:
+ case REAL:
+ case BOOLEAN:
+ sb.append(val.toString());
+ break;
+ case TEXT:
+ sb.append("'").append(val.toString()).append("'");
+ break;
+ case DATE:
+ sb.append("'" + val.toString() + "'");
+ break;
+ default:
+ throw new UnsupportedFilterException("unsupported column type for filtering : " + column.columnTypeCode());
+ }
+
+ }
+ return sb.toString();
+ } catch (UnsupportedFilterException ex) {
+ // Do not throw exception, impose no constraints instead
+ return null;
+ }
+ }
+
+
+ /**
+ * Unsupported filter exception class.
+ * Thrown when the filter string passed to constructor cannot be parsed
+ */
+ private static class UnsupportedFilterException extends Exception {
+ UnsupportedFilterException(String message) {
+ super(message);
+ }
+ }
+
+ // PXF InputData
+ private InputData inputData;
+
+ /**
+ * Parses PXF {@link InputData} 'FilterObject'
+ * Only 'HDOP_AND' is supported as a 'LogicalOperation' at the moment
+ *
+ * @param filter 'FilterObject' to parse
+ * @param returnList a list of 'BasicFilter' to append 'BasicFilter' to
+ *
+ * @throws UnsupportedFilterException if the provided filter string or 'LogicalOperation' is incorrect
+ */
+ private List<BasicFilter> convertBasicFilterList(Object filter, List<BasicFilter> returnList) throws UnsupportedFilterException {
+ if (returnList == null) {
+ returnList = new ArrayList<>();
+ }
+
+ if (filter instanceof BasicFilter) {
+ returnList.add((BasicFilter) filter);
+ return returnList;
+ }
+
+ LogicalFilter lfilter = (LogicalFilter) filter;
+ if (lfilter.getOperator() != FilterParser.LogicalOperation.HDOP_AND) {
+ throw new UnsupportedFilterException("unsupported LogicalOperation : " + lfilter.getOperator());
+ }
+
+ for (Object f : lfilter.getFilterList()) {
+ returnList = convertBasicFilterList(f, returnList);
+ }
+
+ return returnList;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgniteAccessorTest.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgniteAccessorTest.java b/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgniteAccessorTest.java
new file mode 100644
index 0000000..437e8ad
--- /dev/null
+++ b/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgniteAccessorTest.java
@@ -0,0 +1,130 @@
+package org.apache.hawq.pxf.plugins.ignite;
+
+/*
+ * 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.
+ */
+
+import org.apache.hawq.pxf.api.OneRow;
+import org.apache.hawq.pxf.api.io.DataType;
+import org.apache.hawq.pxf.api.utilities.ColumnDescriptor;
+import org.apache.hawq.pxf.api.utilities.InputData;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+
+import org.junit.Test;
+import org.junit.Before;
+import org.junit.runner.RunWith;
+import static org.junit.Assert.*;
+
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import static org.mockito.Matchers.anyString;
+
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import static org.powermock.api.support.membermodification.MemberMatcher.method;
+
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({IgniteAccessor.class})
+public class IgniteAccessorTest {
+ @Before
+ public void prepareAccessorTest() throws Exception {
+ inputData = Mockito.mock(InputData.class);
+
+ Mockito.when(inputData.getDataSource()).thenReturn("TableTest");
+
+ columns.add(new ColumnDescriptor("id", DataType.INTEGER.getOID(), 0, "int4", null));
+ columns.add(new ColumnDescriptor("name", DataType.TEXT.getOID(), 1, "text", null));
+ columns.add(new ColumnDescriptor("birthday", DataType.DATE.getOID(), 2, "date", null));
+ columns.add(new ColumnDescriptor("key", DataType.BYTEA.getOID(), 3, "bytea", null));
+ Mockito.when(inputData.getTupleDescription()).thenReturn(columns);
+ Mockito.when(inputData.getColumn(0)).thenReturn(columns.get(0));
+ Mockito.when(inputData.getColumn(1)).thenReturn(columns.get(1));
+ Mockito.when(inputData.getColumn(2)).thenReturn(columns.get(2));
+ Mockito.when(inputData.getColumn(3)).thenReturn(columns.get(3));
+ }
+
+ @Test
+ public void testReadAccess() throws Exception {
+ IgniteAccessor acc = PowerMockito.spy(new IgniteAccessor(inputData));
+
+ JsonObject correctAnswer = new JsonObject();
+ JsonArray tempArray = new JsonArray();
+ JsonArray tempArray2 = new JsonArray();
+ tempArray2.add(1);
+ tempArray2.add("abcd");
+ tempArray2.add("'2001-01-01'");
+ tempArray2.add("YWJjZA==");
+ tempArray.add(tempArray2);
+ correctAnswer.add("items", tempArray);
+ correctAnswer.addProperty("last", false);
+ correctAnswer.addProperty("queryId", 1);
+
+ ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
+ PowerMockito.doReturn(correctAnswer).when(acc, "sendRestRequest", anyString());
+
+ acc.openForRead();
+ acc.readNextObject();
+ acc.closeForRead();
+
+ PowerMockito.verifyPrivate(acc, Mockito.times(3)).invoke(method(IgniteAccessor.class, "sendRestRequest", String.class)).withArguments(captor.capture());
+
+ List<String> allParams = captor.getAllValues();
+
+ assertEquals(allParams.get(0), "http://127.0.0.1:8080/ignite?cmd=qryfldexe&pageSize=0&qry=SELECT+id%2C+name%2C+birthday%2C+key+FROM+TableTest");
+ assertEquals(allParams.get(1), "http://127.0.0.1:8080/ignite?cmd=qryfetch&pageSize=128&qryId=1");
+ assertEquals(allParams.get(2), "http://127.0.0.1:8080/ignite?cmd=qrycls&qryId=1");
+ }
+
+ @Test
+ public void testWriteAccess() throws Exception {
+ IgniteAccessor acc = PowerMockito.spy(new IgniteAccessor(inputData));
+
+ OneRow insert_row_1 = new OneRow("(1, 'abcd', '2001-01-01', '61626364')");
+ OneRow insert_row_2 = new OneRow("(2, 'abcd', '2001-01-01', '61626364')");
+
+ JsonObject correctAnswer = new JsonObject();
+ correctAnswer.addProperty("queryId", 1);
+
+ ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
+ PowerMockito.doReturn(correctAnswer).when(acc, "sendRestRequest", anyString());
+
+ acc.openForWrite();
+ acc.writeNextObject(insert_row_1);
+ acc.writeNextObject(insert_row_2);
+ acc.closeForWrite();
+
+ PowerMockito.verifyPrivate(acc, Mockito.times(4)).invoke(method(IgniteAccessor.class, "sendRestRequest", String.class)).withArguments(captor.capture());
+
+ List<String> allParams = captor.getAllValues();
+
+ assertEquals(allParams.get(0), "http://127.0.0.1:8080/ignite?cmd=qryfldexe&pageSize=0&qry=INSERT+INTO+TableTest%28id%2C+name%2C+birthday%2C+key%29+VALUES+%281%2C+%27abcd%27%2C+%272001-01-01%27%2C+%2761626364%27%29");
+ assertEquals(allParams.get(1), "http://127.0.0.1:8080/ignite?cmd=qrycls&qryId=1");
+ assertEquals(allParams.get(2), "http://127.0.0.1:8080/ignite?cmd=qryfldexe&pageSize=0&qry=INSERT+INTO+TableTest%28id%2C+name%2C+birthday%2C+key%29+VALUES+%282%2C+%27abcd%27%2C+%272001-01-01%27%2C+%2761626364%27%29");
+ assertEquals(allParams.get(3), "http://127.0.0.1:8080/ignite?cmd=qrycls&qryId=1");
+ }
+
+ private ArrayList<ColumnDescriptor> columns = new ArrayList<>();
+ private InputData inputData = null;
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/3779f14d/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgniteFilterBuilderTest.java
----------------------------------------------------------------------
diff --git a/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgniteFilterBuilderTest.java b/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgniteFilterBuilderTest.java
new file mode 100644
index 0000000..220d0c5
--- /dev/null
+++ b/pxf/pxf-ignite/src/test/java/org/apache/hawq/pxf/plugins/ignite/IgniteFilterBuilderTest.java
@@ -0,0 +1,100 @@
+package org.apache.hawq.pxf.plugins.ignite;
+
+/*
+ * 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.
+ */
+
+import org.apache.hawq.pxf.api.BasicFilter;
+import org.apache.hawq.pxf.api.FilterParser;
+import static org.apache.hawq.pxf.api.FilterParser.Operation.*;
+import org.apache.hawq.pxf.api.FilterParser.LogicalOperation;
+import org.apache.hawq.pxf.api.LogicalFilter;
+
+import org.junit.Test;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+
+public class IgniteFilterBuilderTest {
+ @Test
+ public void parseFilterWithThreeOperations() throws Exception {
+ //orgin sql => col_1>'2008-02-01' and col_1<'2008-12-01' or col_2 > 1200
+ String filterstr = "a1c25s10d2008-02-01o2a1c25s10d2008-12-01o1l0a2c20s4d1200o2l1";
+ IgniteFilterBuilder builder = new IgniteFilterBuilder();
+
+ LogicalFilter filterList = (LogicalFilter) builder.getFilterObject(filterstr);
+ assertEquals(LogicalOperation.HDOP_OR, filterList.getOperator());
+ LogicalFilter l1_left = (LogicalFilter) filterList.getFilterList().get(0);
+ BasicFilter l1_right = (BasicFilter) filterList.getFilterList().get(1);
+
+ //column_2 > 1200
+ assertEquals(2, l1_right.getColumn().index());
+ assertEquals(HDOP_GT, l1_right.getOperation());
+ assertEquals(1200L, l1_right.getConstant().constant());
+
+ assertEquals(LogicalOperation.HDOP_AND, l1_left.getOperator());
+ BasicFilter l2_left = (BasicFilter) l1_left.getFilterList().get(0);
+ BasicFilter l2_right = (BasicFilter) l1_left.getFilterList().get(1);
+
+ //column_1 > '2008-02-01'
+ assertEquals(1, l2_left.getColumn().index());
+ assertEquals(HDOP_GT, l2_left.getOperation());
+ assertEquals("2008-02-01", l2_left.getConstant().constant());
+
+ //column_2 < '2008-12-01'
+ assertEquals(1, l2_right.getColumn().index());
+ assertEquals(HDOP_LT, l2_right.getOperation());
+ assertEquals("2008-12-01", l2_right.getConstant().constant());
+
+ }
+
+ @Test
+ public void parseFilterWithLogicalOperation() throws Exception {
+ WhereSQLBuilder builder = new WhereSQLBuilder(null);
+ LogicalFilter filter = (LogicalFilter) builder.getFilterObject("a1c25s5dfirsto5a2c20s1d2o2l0");
+ assertEquals(LogicalOperation.HDOP_AND, filter.getOperator());
+ assertEquals(2, filter.getFilterList().size());
+ }
+
+ @Test
+ public void parseNestedExpressionWithLogicalOperation() throws Exception {
+ WhereSQLBuilder builder = new WhereSQLBuilder(null);
+ LogicalFilter filter = (LogicalFilter) builder.getFilterObject("a1c25s5dfirsto5a2c20s1d2o2l0a1c20s1d1o1l1");
+ assertEquals(LogicalOperation.HDOP_OR, filter.getOperator());
+ assertEquals(LogicalOperation.HDOP_AND, ((LogicalFilter) filter.getFilterList().get(0)).getOperator());
+ assertEquals(HDOP_LT, ((BasicFilter) filter.getFilterList().get(1)).getOperation());
+ }
+
+ @Test
+ public void parseIsNullExpression() throws Exception {
+ WhereSQLBuilder builder = new WhereSQLBuilder(null);
+ BasicFilter filter = (BasicFilter) builder.getFilterObject("a1o8");
+ assertEquals(FilterParser.Operation.HDOP_IS_NULL, filter.getOperation());
+ assertEquals(1, filter.getColumn().index());
+ assertNull(filter.getConstant());
+ }
+
+ @Test
+ public void parseIsNotNullExpression() throws Exception {
+ WhereSQLBuilder builder = new WhereSQLBuilder(null);
+ BasicFilter filter = (BasicFilter) builder.getFilterObject("a1o9");
+ assertEquals(FilterParser.Operation.HDOP_IS_NOT_NULL, filter.getOperation());
+ assertEquals(1, filter.getColumn().index());
+ assertNull(filter.getConstant());
+ }
+}