Repository: incubator-hawq Updated Branches: refs/heads/master 4d3375516 -> fe33faaba
HAWQ-1542. PXF Demo profile should support write use case. Project: http://git-wip-us.apache.org/repos/asf/incubator-hawq/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-hawq/commit/fe33faab Tree: http://git-wip-us.apache.org/repos/asf/incubator-hawq/tree/fe33faab Diff: http://git-wip-us.apache.org/repos/asf/incubator-hawq/diff/fe33faab Branch: refs/heads/master Commit: fe33faaba9e4f0422240014c415d1b6a999db5f8 Parents: 4d33755 Author: Alexander Denissov <[email protected]> Authored: Thu Oct 26 11:44:10 2017 -0700 Committer: Alexander Denissov <[email protected]> Committed: Tue Oct 31 10:34:26 2017 -0700 ---------------------------------------------------------------------- .../java/org/apache/hawq/pxf/api/OneRow.java | 9 ++ .../api/examples/DemoFileWritableAccessor.java | 102 +++++++++++++++++++ .../hawq/pxf/api/examples/DemoTextResolver.java | 26 ++++- .../apache/hawq/pxf/api/DemoResolverTest.java | 48 ++++++++- 4 files changed, 178 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/fe33faab/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/OneRow.java ---------------------------------------------------------------------- diff --git a/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/OneRow.java b/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/OneRow.java index 263d360..85a5edf 100755 --- a/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/OneRow.java +++ b/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/OneRow.java @@ -43,6 +43,15 @@ public class OneRow { this.data = data; } + /** + * Constructs a OneRow without a key value + * + * @param data the actual record + */ + public OneRow(Object data) { + this.data = data; + } + public void setKey(Object key) { this.key = key; } http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/fe33faab/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/examples/DemoFileWritableAccessor.java ---------------------------------------------------------------------- diff --git a/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/examples/DemoFileWritableAccessor.java b/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/examples/DemoFileWritableAccessor.java new file mode 100644 index 0000000..65376da --- /dev/null +++ b/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/examples/DemoFileWritableAccessor.java @@ -0,0 +1,102 @@ +package org.apache.hawq.pxf.api.examples; + +/* + * 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.WriteAccessor; +import org.apache.hawq.pxf.api.utilities.InputData; +import org.apache.hawq.pxf.api.utilities.Plugin; + +import java.io.BufferedOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.nio.file.FileSystems; +import java.nio.file.Files; +import java.nio.file.Path; + +/** + * PXF Accessor for writing text data into a local file. + * + * Demo implementation. + */ + +public class DemoFileWritableAccessor extends Plugin implements WriteAccessor { + + private OutputStream out; + + /** + * Constructs a DemoFileWritableAccessor. + * + * @param input all input parameters coming from the client request + */ + public DemoFileWritableAccessor(InputData input) { + super(input); + } + + /** + * Opens the resource for write. + * + * @return true if the resource is successfully opened + * @throws Exception if opening the resource failed + */ + @Override + public boolean openForWrite() throws Exception { + String fileName = inputData.getDataSource(); + + Path file = FileSystems.getDefault().getPath(fileName); + if (Files.exists(file)) { + throw new IOException("File " + file.toString() + " already exists."); + } + + Path parent = file.getParent(); + if (Files.notExists(parent)) { + Files.createDirectories(parent); + } + + out = new BufferedOutputStream(Files.newOutputStream(file)); + return true; + } + + /** + * Writes the next object. + * + * @param onerow the object to be written + * @return true if the write succeeded + * @throws Exception writing to the resource failed + */ + @Override + public boolean writeNextObject(OneRow onerow) throws Exception { + out.write((byte[]) onerow.getData()); + return true; + } + + /** + * Closes the resource for write. + * + * @throws Exception if closing the resource failed + */ + @Override + public void closeForWrite() throws Exception { + if (out != null) { + out.flush(); + out.close(); + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/fe33faab/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/examples/DemoTextResolver.java ---------------------------------------------------------------------- diff --git a/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/examples/DemoTextResolver.java b/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/examples/DemoTextResolver.java index ff23daf..36b0427 100644 --- a/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/examples/DemoTextResolver.java +++ b/pxf/pxf-api/src/main/java/org/apache/hawq/pxf/api/examples/DemoTextResolver.java @@ -22,21 +22,22 @@ package org.apache.hawq.pxf.api.examples; import org.apache.hawq.pxf.api.OneField; import org.apache.hawq.pxf.api.OneRow; import org.apache.hawq.pxf.api.ReadResolver; +import org.apache.hawq.pxf.api.WriteResolver; import org.apache.hawq.pxf.api.utilities.InputData; import org.apache.hawq.pxf.api.utilities.Plugin; import java.util.LinkedList; import java.util.List; -import static org.apache.hawq.pxf.api.io.DataType.INTEGER; import static org.apache.hawq.pxf.api.io.DataType.VARCHAR; /** - * Class that defines the deserializtion of one record brought from the external input data. + * Class that defines the serialization / deserialization of one record brought from the external input data. * * Demo implementation of resolver that returns text format */ -public class DemoTextResolver extends Plugin implements ReadResolver { +public class DemoTextResolver extends Plugin implements ReadResolver, WriteResolver { + /** * Constructs the DemoResolver * @@ -60,4 +61,23 @@ public class DemoTextResolver extends Plugin implements ReadResolver { output.add(new OneField(VARCHAR.getOID(), data)); return output; } + + /** + * Creates a OneRow object from the singleton list. + * + * @param record list of {@link OneField} + * @return the constructed {@link OneRow} + * @throws Exception if constructing a row from the fields failed + */ + @Override + public OneRow setFields(List<OneField> record) throws Exception { + // text row data is passed as a single field + if (record == null || record.size() != 1) { + throw new Exception("Unexpected record format, expected 1 field, found " + + (record == null ? 0 : record.size())); + } + byte[] value = (byte[]) record.get(0).val; + // empty array means the end of input stream, return null to stop iterations + return value.length == 0 ? null : new OneRow(value); + } } http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/fe33faab/pxf/pxf-api/src/test/java/org/apache/hawq/pxf/api/DemoResolverTest.java ---------------------------------------------------------------------- diff --git a/pxf/pxf-api/src/test/java/org/apache/hawq/pxf/api/DemoResolverTest.java b/pxf/pxf-api/src/test/java/org/apache/hawq/pxf/api/DemoResolverTest.java index dd22c87..5417981 100755 --- a/pxf/pxf-api/src/test/java/org/apache/hawq/pxf/api/DemoResolverTest.java +++ b/pxf/pxf-api/src/test/java/org/apache/hawq/pxf/api/DemoResolverTest.java @@ -31,8 +31,11 @@ import org.mockito.Mock; import org.powermock.core.classloader.annotations.PrepareForTest; import org.powermock.modules.junit4.PowerMockRunner; +import java.util.Arrays; +import java.util.Collections; import java.util.List; +import static org.apache.hawq.pxf.api.io.DataType.VARCHAR; import static org.junit.Assert.*; @RunWith(PowerMockRunner.class) @@ -40,20 +43,24 @@ import static org.junit.Assert.*; public class DemoResolverTest { + private static final String DATA = "value1,value2"; + @Mock InputData inputData; DemoResolver customResolver; DemoTextResolver textResolver; OneRow row; + OneField field; @Before public void setup() throws Exception { customResolver = new DemoResolver(inputData); textResolver = new DemoTextResolver(inputData); - row = new OneRow("0.0","value1,value2"); + row = new OneRow("0.0", DATA); + field = new OneField(VARCHAR.getOID(), DATA.getBytes()); } @Test - public void testCustomData() throws Exception { + public void testGetCustomData() throws Exception { List<OneField> output = customResolver.getFields(row); assertEquals("value1", output.get(0).toString()); @@ -61,10 +68,43 @@ public class DemoResolverTest { } @Test - public void testTextData() throws Exception { + public void testGetTextData() throws Exception { List<OneField> output = textResolver.getFields(row); - assertEquals("value1,value2", output.get(0).toString()); + assertEquals(DATA, output.get(0).toString()); + } + + @Test + public void testSetTextData() throws Exception { + + OneRow output = textResolver.setFields(Collections.singletonList(field)); + assertArrayEquals(DATA.getBytes(), (byte[]) output.getData()); + } + + @Test + public void testSetEmptyTextData() throws Exception { + + OneField field = new OneField(VARCHAR.getOID(), new byte[]{}); + OneRow output = textResolver.setFields(Collections.singletonList(field)); + assertNull(output); + } + + @Test(expected = Exception.class) + public void testSetTextDataNullInput() throws Exception { + + textResolver.setFields(null); + } + + @Test(expected = Exception.class) + public void testSetTextDataEmptyInput() throws Exception { + textResolver.setFields(Collections.emptyList()); } + + @Test(expected = Exception.class) + public void testSetTextDataManyElements() throws Exception { + + textResolver.setFields(Arrays.asList(field, field)); + } + }
