You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ma...@apache.org on 2015/12/11 03:44:23 UTC
[45/52] [abbrv] phoenix git commit: PHOENIX-2481 JSON bulkload tool
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
index eae58ad..05fbab2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
@@ -66,9 +66,9 @@ import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
-import org.apache.phoenix.mapreduce.CsvBulkLoadTool.TargetTableRef;
-import org.apache.phoenix.mapreduce.CsvBulkLoadTool.TargetTableRefFunctions;
-import org.apache.phoenix.mapreduce.bulkload.CsvTableRowkeyPair;
+import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
+import org.apache.phoenix.mapreduce.bulkload.TargetTableRef;
+import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -81,7 +81,7 @@ import com.google.common.collect.Sets;
* It has been adapted from {#link HFileOutputFormat2} but differs from the fact it creates
* HFiles for multiple tables.
*/
-public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair, Cell> {
+public class MultiHfileOutputFormat extends FileOutputFormat<TableRowkeyPair, Cell> {
private static final Logger LOG = LoggerFactory.getLogger(MultiHfileOutputFormat.class);
@@ -101,7 +101,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
private static final String AT_DELIMITER = "@";
@Override
- public RecordWriter<CsvTableRowkeyPair, Cell> getRecordWriter(TaskAttemptContext context)
+ public RecordWriter<TableRowkeyPair, Cell> getRecordWriter(TaskAttemptContext context)
throws IOException, InterruptedException {
return createRecordWriter(context);
}
@@ -112,7 +112,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
* @return
* @throws IOException
*/
- static <V extends Cell> RecordWriter<CsvTableRowkeyPair, V> createRecordWriter(final TaskAttemptContext context)
+ static <V extends Cell> RecordWriter<TableRowkeyPair, V> createRecordWriter(final TaskAttemptContext context)
throws IOException {
// Get the path of the temporary output file
final Path outputPath = FileOutputFormat.getOutputPath(context);
@@ -130,7 +130,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
final boolean compactionExclude = conf.getBoolean(
"hbase.mapreduce.hfileoutputformat.compaction.exclude", false);
- return new RecordWriter<CsvTableRowkeyPair, V>() {
+ return new RecordWriter<TableRowkeyPair, V>() {
// Map of families to writers and how much has been output on the writer.
private final Map<byte [], WriterLength> writers =
new TreeMap<byte [], WriterLength>(Bytes.BYTES_COMPARATOR);
@@ -139,7 +139,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
private boolean rollRequested = false;
@Override
- public void write(CsvTableRowkeyPair row, V cell)
+ public void write(TableRowkeyPair row, V cell)
throws IOException {
KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
// null input == user explicitly wants to flush
@@ -450,7 +450,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
* Configure <code>job</code> with a TotalOrderPartitioner, partitioning against
* <code>splitPoints</code>. Cleans up the partitions file after job exists.
*/
- static void configurePartitioner(Job job, Set<CsvTableRowkeyPair> tablesStartKeys)
+ static void configurePartitioner(Job job, Set<TableRowkeyPair> tablesStartKeys)
throws IOException {
Configuration conf = job.getConfiguration();
@@ -467,7 +467,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
}
private static void writePartitions(Configuration conf, Path partitionsPath,
- Set<CsvTableRowkeyPair> tablesStartKeys) throws IOException {
+ Set<TableRowkeyPair> tablesStartKeys) throws IOException {
LOG.info("Writing partition information to " + partitionsPath);
if (tablesStartKeys.isEmpty()) {
@@ -478,9 +478,9 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
// have keys < the first region (which has an empty start key)
// so we need to remove it. Otherwise we would end up with an
// empty reducer with index 0
- TreeSet<CsvTableRowkeyPair> sorted = new TreeSet<CsvTableRowkeyPair>(tablesStartKeys);
+ TreeSet<TableRowkeyPair> sorted = new TreeSet<TableRowkeyPair>(tablesStartKeys);
- CsvTableRowkeyPair first = sorted.first();
+ TableRowkeyPair first = sorted.first();
if (!first.getRowkey().equals(HConstants.EMPTY_BYTE_ARRAY)) {
throw new IllegalArgumentException(
"First region of table should have empty start key. Instead has: "
@@ -491,11 +491,11 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
// Write the actual file
FileSystem fs = partitionsPath.getFileSystem(conf);
SequenceFile.Writer writer = SequenceFile.createWriter(
- fs, conf, partitionsPath, CsvTableRowkeyPair.class,
+ fs, conf, partitionsPath, TableRowkeyPair.class,
NullWritable.class);
try {
- for (CsvTableRowkeyPair startKey : sorted) {
+ for (TableRowkeyPair startKey : sorted) {
writer.append(startKey, NullWritable.get());
}
} finally {
@@ -658,11 +658,11 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
KeyValueSerialization.class.getName());
// tableStartKeys for all tables.
- Set<CsvTableRowkeyPair> tablesStartKeys = Sets.newTreeSet();
+ Set<TableRowkeyPair> tablesStartKeys = Sets.newTreeSet();
for(TargetTableRef table : tablesToBeLoaded) {
final String tableName = table.getPhysicalName();
try(HTable htable = new HTable(conf,tableName);){
- Set<CsvTableRowkeyPair> startKeys = getRegionStartKeys(tableName , htable.getRegionLocator());
+ Set<TableRowkeyPair> startKeys = getRegionStartKeys(tableName , htable.getRegionLocator());
tablesStartKeys.addAll(startKeys);
String compressionConfig = configureCompression(htable.getTableDescriptor());
String bloomTypeConfig = configureBloomType(htable.getTableDescriptor());
@@ -704,12 +704,12 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
* Return the start keys of all of the regions in this table,
* as a list of ImmutableBytesWritable.
*/
- private static Set<CsvTableRowkeyPair> getRegionStartKeys(String tableName , RegionLocator table) throws IOException {
+ private static Set<TableRowkeyPair> getRegionStartKeys(String tableName , RegionLocator table) throws IOException {
byte[][] byteKeys = table.getStartKeys();
- Set<CsvTableRowkeyPair> ret = new TreeSet<CsvTableRowkeyPair>();
+ Set<TableRowkeyPair> ret = new TreeSet<TableRowkeyPair>();
for (byte[] byteKey : byteKeys) {
// phoenix-2216: start : passing the table name and startkey
- ret.add(new CsvTableRowkeyPair(tableName, new ImmutableBytesWritable(byteKey)));
+ ret.add(new TableRowkeyPair(tableName, new ImmutableBytesWritable(byteKey)));
}
return ret;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/CsvTableRowkeyPair.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/CsvTableRowkeyPair.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/CsvTableRowkeyPair.java
deleted file mode 100644
index 3ae74b6..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/CsvTableRowkeyPair.java
+++ /dev/null
@@ -1,139 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.phoenix.mapreduce.bulkload;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.io.BytesWritable;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.io.WritableComparator;
-import org.apache.hadoop.io.WritableUtils;
-
-import com.google.common.base.Preconditions;
-
-
-/**
- * A WritableComparable to hold the table name and the rowkey.
- *
- */
-public class CsvTableRowkeyPair implements WritableComparable<CsvTableRowkeyPair> {
-
- /* The qualified table name */
- private String tableName;
-
- /* The rowkey for the record */
- private ImmutableBytesWritable rowkey;
-
- /**
- * Default constructor
- */
- public CsvTableRowkeyPair() {
- super();
- }
-
- /**
- * @param tableName
- * @param rowkey
- */
- public CsvTableRowkeyPair(String tableName, ImmutableBytesWritable rowkey) {
- super();
- Preconditions.checkNotNull(tableName);
- Preconditions.checkNotNull(rowkey);
- this.tableName = tableName;
- this.rowkey = rowkey;
- }
-
- public String getTableName() {
- return tableName;
- }
-
- public void setTableName(String tableName) {
- this.tableName = tableName;
- }
-
- public ImmutableBytesWritable getRowkey() {
- return rowkey;
- }
-
- public void setRowkey(ImmutableBytesWritable rowkey) {
- this.rowkey = rowkey;
- }
-
- @Override
- public void readFields(DataInput input) throws IOException {
- tableName = WritableUtils.readString(input);
- rowkey = new ImmutableBytesWritable();
- rowkey.readFields(input);
- }
-
- @Override
- public void write(DataOutput output) throws IOException {
- WritableUtils.writeString(output,tableName);
- rowkey.write(output);
- }
-
- @Override
- public int compareTo(CsvTableRowkeyPair other) {
- String otherTableName = other.getTableName();
- if(this.tableName.equals(otherTableName)) {
- return this.rowkey.compareTo(other.getRowkey());
- } else {
- return this.tableName.compareTo(otherTableName);
- }
- }
-
- /** Comparator optimized for <code>CsvTableRowkeyPair</code>. */
- public static class Comparator extends WritableComparator {
- private BytesWritable.Comparator comparator = new BytesWritable.Comparator();
-
- public Comparator() {
- super(CsvTableRowkeyPair.class);
- }
-
- @Override
- public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
- try {
- int vintL1 = WritableUtils.decodeVIntSize(b1[s1]);
- int vintL2 = WritableUtils.decodeVIntSize(b2[s2]);
- int strL1 = readVInt(b1, s1);
- int strL2 = readVInt(b2, s2);
- int cmp = compareBytes(b1, s1 + vintL1, strL1, b2, s2 + vintL2, strL2);
- if (cmp != 0) {
- return cmp;
- }
- int vintL3 = WritableUtils.decodeVIntSize(b1[s1 + vintL1 + strL1]);
- int vintL4 = WritableUtils.decodeVIntSize(b2[s2 + vintL2 + strL2]);
- int strL3 = readVInt(b1, s1 + vintL1 + strL1);
- int strL4 = readVInt(b2, s2 + vintL2 + strL2);
- return comparator.compare(b1, s1 + vintL1 + strL1 + vintL3, strL3, b2, s2
- + vintL2 + strL2 + vintL4, strL4);
-
- } catch(Exception ex) {
- throw new IllegalArgumentException(ex);
- }
- }
- }
-
- static {
- WritableComparator.define(CsvTableRowkeyPair.class, new Comparator());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TableRowkeyPair.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TableRowkeyPair.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TableRowkeyPair.java
new file mode 100644
index 0000000..412226f
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TableRowkeyPair.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.mapreduce.bulkload;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableComparator;
+import org.apache.hadoop.io.WritableUtils;
+
+import com.google.common.base.Preconditions;
+
+
+/**
+ * A WritableComparable to hold the table name and the rowkey.
+ */
+public class TableRowkeyPair implements WritableComparable<TableRowkeyPair> {
+
+ /* The qualified table name */
+ private String tableName;
+
+ /* The rowkey for the record */
+ private ImmutableBytesWritable rowkey;
+
+ /**
+ * Default constructor
+ */
+ public TableRowkeyPair() {
+ super();
+ }
+
+ public TableRowkeyPair(String tableName, ImmutableBytesWritable rowkey) {
+ super();
+ Preconditions.checkNotNull(tableName);
+ Preconditions.checkNotNull(rowkey);
+ this.tableName = tableName;
+ this.rowkey = rowkey;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public void setTableName(String tableName) {
+ this.tableName = tableName;
+ }
+
+ public ImmutableBytesWritable getRowkey() {
+ return rowkey;
+ }
+
+ public void setRowkey(ImmutableBytesWritable rowkey) {
+ this.rowkey = rowkey;
+ }
+
+ @Override
+ public void readFields(DataInput input) throws IOException {
+ tableName = WritableUtils.readString(input);
+ rowkey = new ImmutableBytesWritable();
+ rowkey.readFields(input);
+ }
+
+ @Override
+ public void write(DataOutput output) throws IOException {
+ WritableUtils.writeString(output,tableName);
+ rowkey.write(output);
+ }
+
+ @Override
+ public int compareTo(TableRowkeyPair other) {
+ String otherTableName = other.getTableName();
+ if(this.tableName.equals(otherTableName)) {
+ return this.rowkey.compareTo(other.getRowkey());
+ } else {
+ return this.tableName.compareTo(otherTableName);
+ }
+ }
+
+ /** Comparator optimized for <code>TableRowkeyPair</code>. */
+ public static class Comparator extends WritableComparator {
+ private BytesWritable.Comparator comparator = new BytesWritable.Comparator();
+
+ public Comparator() {
+ super(TableRowkeyPair.class);
+ }
+
+ @Override
+ public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
+ try {
+ int vintL1 = WritableUtils.decodeVIntSize(b1[s1]);
+ int vintL2 = WritableUtils.decodeVIntSize(b2[s2]);
+ int strL1 = readVInt(b1, s1);
+ int strL2 = readVInt(b2, s2);
+ int cmp = compareBytes(b1, s1 + vintL1, strL1, b2, s2 + vintL2, strL2);
+ if (cmp != 0) {
+ return cmp;
+ }
+ int vintL3 = WritableUtils.decodeVIntSize(b1[s1 + vintL1 + strL1]);
+ int vintL4 = WritableUtils.decodeVIntSize(b2[s2 + vintL2 + strL2]);
+ int strL3 = readVInt(b1, s1 + vintL1 + strL1);
+ int strL4 = readVInt(b2, s2 + vintL2 + strL2);
+ return comparator.compare(b1, s1 + vintL1 + strL1 + vintL3, strL3, b2, s2
+ + vintL2 + strL2 + vintL4, strL4);
+
+ } catch(Exception ex) {
+ throw new IllegalArgumentException(ex);
+ }
+ }
+ }
+
+ static {
+ WritableComparator.define(TableRowkeyPair.class, new Comparator());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRef.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRef.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRef.java
new file mode 100644
index 0000000..1a846f9
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRef.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.mapreduce.bulkload;
+
+import java.util.Map;
+
+import org.codehaus.jackson.annotate.JsonCreator;
+import org.codehaus.jackson.annotate.JsonProperty;
+
+import com.google.common.collect.Maps;
+
+/**
+ * Represents the logical and physical name of a single table to which data is to be loaded.
+ *
+ * This class exists to allow for the difference between HBase physical table names and
+ * Phoenix logical table names.
+ */
+public class TargetTableRef {
+
+ @JsonProperty
+ private final String logicalName;
+
+ @JsonProperty
+ private final String physicalName;
+
+ @JsonProperty
+ private Map<String,String> configuration = Maps.newHashMap();
+
+ public TargetTableRef(String name) {
+ this(name, name);
+ }
+
+ @JsonCreator
+ private TargetTableRef(@JsonProperty("logicalName") String logicalName,
+ @JsonProperty("physicalName") String physicalName) {
+ this.logicalName = logicalName;
+ this.physicalName = physicalName;
+ }
+
+ public String getLogicalName() {
+ return logicalName;
+ }
+
+ public String getPhysicalName() {
+ return physicalName;
+ }
+
+ public Map<String, String> getConfiguration() {
+ return configuration;
+ }
+
+ public void setConfiguration(Map<String, String> configuration) {
+ this.configuration = configuration;
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
new file mode 100644
index 0000000..d786842
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.mapreduce.bulkload;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.codehaus.jackson.map.ObjectMapper;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+
+/**
+ * Utility functions to get/put json.
+ *
+ */
+public class TargetTableRefFunctions {
+
+ public static Function<TargetTableRef,String> TO_JSON = new Function<TargetTableRef,String>() {
+
+ @Override
+ public String apply(TargetTableRef input) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.writeValueAsString(input);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
+
+ public static Function<String,TargetTableRef> FROM_JSON = new Function<String,TargetTableRef>() {
+
+ @Override
+ public TargetTableRef apply(String json) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.readValue(json, TargetTableRef.class);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
+
+ public static Function<List<TargetTableRef>,String> NAMES_TO_JSON = new Function<List<TargetTableRef>,String>() {
+
+ @Override
+ public String apply(List<TargetTableRef> input) {
+ try {
+ List<String> tableNames = Lists.newArrayListWithCapacity(input.size());
+ for(TargetTableRef table : input) {
+ tableNames.add(table.getPhysicalName());
+ }
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.writeValueAsString(tableNames);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
+
+ public static Function<String,List<String>> NAMES_FROM_JSON = new Function<String,List<String>>() {
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public List<String> apply(String json) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.readValue(json, ArrayList.class);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
+ }
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
index 8bf786b..9e29fba 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
@@ -23,13 +23,6 @@ import java.sql.SQLException;
import java.util.List;
import java.util.Map;
-import com.google.common.base.Function;
-import com.google.common.base.Joiner;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Splitter;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -41,13 +34,17 @@ import org.apache.hadoop.mapreduce.lib.db.DBInputFormat.NullDBWritable;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.phoenix.jdbc.PhoenixConnection;
-import org.apache.phoenix.mapreduce.CsvToKeyValueMapper.DefaultImportPreUpsertKeyValueProcessor;
+import org.apache.phoenix.mapreduce.FormatToKeyValueMapper;
import org.apache.phoenix.mapreduce.ImportPreUpsertKeyValueProcessor;
import org.apache.phoenix.mapreduce.PhoenixInputFormat;
import org.apache.phoenix.util.ColumnInfo;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.QueryUtil;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
import static org.apache.commons.lang.StringUtils.isNotEmpty;
/**
@@ -419,7 +416,7 @@ public final class PhoenixConfigurationUtil {
Class<? extends ImportPreUpsertKeyValueProcessor> processorClass = null;
try {
processorClass = conf.getClass(
- UPSERT_HOOK_CLASS_CONFKEY, DefaultImportPreUpsertKeyValueProcessor.class,
+ UPSERT_HOOK_CLASS_CONFKEY, FormatToKeyValueMapper.DefaultImportPreUpsertKeyValueProcessor.class,
ImportPreUpsertKeyValueProcessor.class);
} catch (Exception e) {
throw new IllegalStateException("Couldn't load upsert hook class", e);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/CSVCommonsLoader.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/CSVCommonsLoader.java b/phoenix-core/src/main/java/org/apache/phoenix/util/CSVCommonsLoader.java
index b8b284a..cdd9d7b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/CSVCommonsLoader.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/CSVCommonsLoader.java
@@ -17,36 +17,25 @@
*/
package org.apache.phoenix.util;
-import com.google.common.base.Charsets;
-import com.google.common.base.Joiner;
-import com.google.common.base.Throwables;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
+import java.io.File;
+import java.io.Reader;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
-import org.apache.phoenix.exception.SQLExceptionCode;
-import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.jdbc.PhoenixConnection;
-import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.util.csv.CsvUpsertExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.File;
-import java.io.Reader;
-import java.sql.Connection;
-import java.sql.DatabaseMetaData;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
+import com.google.common.base.Charsets;
+import com.google.common.base.Joiner;
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableMap;
/***
* Upserts CSV data using Phoenix JDBC connection
@@ -222,7 +211,7 @@ public class CSVCommonsLoader {
long start = System.currentTimeMillis();
CsvUpsertListener upsertListener = new CsvUpsertListener(conn,
conn.getMutateBatchSize(), isStrict);
- CsvUpsertExecutor csvUpsertExecutor = CsvUpsertExecutor.create(conn, tableName,
+ CsvUpsertExecutor csvUpsertExecutor = new CsvUpsertExecutor(conn, tableName,
columnInfoList, upsertListener, arrayElementSeparator);
csvUpsertExecutor.execute(csvParser);
@@ -267,131 +256,10 @@ public class CSVCommonsLoader {
default:
throw new IllegalStateException("parser has unknown column source.");
}
- return generateColumnInfo(conn, tableName, columns, isStrict);
- }
-
- /**
- * Get list of ColumnInfos that contain Column Name and its associated
- * PDataType for an import. The supplied list of columns can be null -- if it is non-null,
- * it represents a user-supplied list of columns to be imported.
- *
- * @param conn Phoenix connection from which metadata will be read
- * @param tableName Phoenix table name whose columns are to be checked. Can include a schema
- * name
- * @param columns user-supplied list of import columns, can be null
- * @param strict if true, an exception will be thrown if unknown columns are supplied
- */
- public static List<ColumnInfo> generateColumnInfo(Connection conn,
- String tableName, List<String> columns, boolean strict)
- throws SQLException {
- Map<String, Integer> columnNameToTypeMap = Maps.newLinkedHashMap();
- Set<String> ambiguousColumnNames = new HashSet<String>();
- Map<String, Integer> fullColumnNameToTypeMap = Maps.newLinkedHashMap();
- DatabaseMetaData dbmd = conn.getMetaData();
- int unfoundColumnCount = 0;
- // TODO: escape wildcard characters here because we don't want that
- // behavior here
- String escapedTableName = StringUtil.escapeLike(tableName);
- String[] schemaAndTable = escapedTableName.split("\\.");
- ResultSet rs = null;
- try {
- rs = dbmd.getColumns(null, (schemaAndTable.length == 1 ? ""
- : schemaAndTable[0]),
- (schemaAndTable.length == 1 ? escapedTableName
- : schemaAndTable[1]), null);
- while (rs.next()) {
- String colName = rs.getString(QueryUtil.COLUMN_NAME_POSITION);
- String colFam = rs.getString(QueryUtil.COLUMN_FAMILY_POSITION);
-
- // use family qualifier, if available, otherwise, use column name
- String fullColumn = (colFam==null?colName:String.format("%s.%s",colFam,colName));
- String sqlTypeName = rs.getString(QueryUtil.DATA_TYPE_NAME_POSITION);
-
- // allow for both bare and family qualified names.
- if (columnNameToTypeMap.keySet().contains(colName)) {
- ambiguousColumnNames.add(colName);
- }
- columnNameToTypeMap.put(
- colName,
- PDataType.fromSqlTypeName(sqlTypeName).getSqlType());
- fullColumnNameToTypeMap.put(
- fullColumn,
- PDataType.fromSqlTypeName(sqlTypeName).getSqlType());
- }
- if (columnNameToTypeMap.isEmpty()) {
- throw new IllegalArgumentException("Table " + tableName + " not found");
- }
- } finally {
- if (rs != null) {
- rs.close();
- }
- }
- List<ColumnInfo> columnInfoList = Lists.newArrayList();
- Set<String> unresolvedColumnNames = new TreeSet<String>();
- if (columns == null) {
- // use family qualified names by default, if no columns are specified.
- for (Map.Entry<String, Integer> entry : fullColumnNameToTypeMap
- .entrySet()) {
- columnInfoList.add(new ColumnInfo(entry.getKey(), entry.getValue()));
- }
- } else {
- // Leave "null" as indication to skip b/c it doesn't exist
- for (int i = 0; i < columns.size(); i++) {
- String columnName = columns.get(i).trim();
- Integer sqlType = null;
- if (fullColumnNameToTypeMap.containsKey(columnName)) {
- sqlType = fullColumnNameToTypeMap.get(columnName);
- } else if (columnNameToTypeMap.containsKey(columnName)) {
- if (ambiguousColumnNames.contains(columnName)) {
- unresolvedColumnNames.add(columnName);
- }
- // fall back to bare column name.
- sqlType = columnNameToTypeMap.get(columnName);
- }
- if (unresolvedColumnNames.size()>0) {
- StringBuilder exceptionMessage = new StringBuilder();
- boolean first = true;
- exceptionMessage.append("Unable to resolve these column names to a single column family:\n");
- for (String col : unresolvedColumnNames) {
- if (first) first = false;
- else exceptionMessage.append(",");
- exceptionMessage.append(col);
- }
- exceptionMessage.append("\nAvailable columns with column families:\n");
- first = true;
- for (String col : fullColumnNameToTypeMap.keySet()) {
- if (first) first = false;
- else exceptionMessage.append(",");
- exceptionMessage.append(col);
- }
- throw new SQLException(exceptionMessage.toString());
- }
-
- if (sqlType == null) {
- if (strict) {
- throw new SQLExceptionInfo.Builder(
- SQLExceptionCode.COLUMN_NOT_FOUND)
- .setColumnName(columnName)
- .setTableName(tableName).build()
- .buildException();
- }
- unfoundColumnCount++;
- } else {
- columnInfoList.add(new ColumnInfo(columnName, sqlType));
- }
- }
- if (unfoundColumnCount == columns.size()) {
- throw new SQLExceptionInfo.Builder(
- SQLExceptionCode.COLUMN_NOT_FOUND)
- .setColumnName(
- Arrays.toString(columns.toArray(new String[0])))
- .setTableName(tableName).build().buildException();
- }
- }
- return columnInfoList;
+ return SchemaUtil.generateColumnInfo(conn, tableName, columns, isStrict);
}
- static class CsvUpsertListener implements CsvUpsertExecutor.UpsertListener {
+ static class CsvUpsertListener implements UpsertExecutor.UpsertListener<CSVRecord> {
private final PhoenixConnection conn;
private final int upsertBatchSize;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
index 1693600..cb8aced 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
@@ -17,21 +17,20 @@
*/
package org.apache.phoenix.util;
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
-import static com.google.common.base.Strings.isNullOrEmpty;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_NAME_BYTES;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_STATS_NAME_BYTES;
-
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-
+import java.util.Set;
+import java.util.TreeSet;
import javax.annotation.Nullable;
import org.apache.hadoop.hbase.KeyValue;
@@ -64,6 +63,15 @@ import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.schema.types.PVarchar;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_NAME_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_STATS_NAME_BYTES;
/**
*
@@ -757,4 +765,125 @@ public class SchemaUtil {
public static byte getSeparatorByte(boolean rowKeyOrderOptimizable, boolean isNullValue, Expression e) {
return getSeparatorByte(rowKeyOrderOptimizable, isNullValue, e.getSortOrder());
}
+
+ /**
+ * Get list of ColumnInfos that contain Column Name and its associated
+ * PDataType for an import. The supplied list of columns can be null -- if it is non-null,
+ * it represents a user-supplied list of columns to be imported.
+ *
+ * @param conn Phoenix connection from which metadata will be read
+ * @param tableName Phoenix table name whose columns are to be checked. Can include a schema
+ * name
+ * @param columns user-supplied list of import columns, can be null
+ * @param strict if true, an exception will be thrown if unknown columns are supplied
+ */
+ public static List<ColumnInfo> generateColumnInfo(Connection conn,
+ String tableName, List<String> columns, boolean strict)
+ throws SQLException {
+ Map<String, Integer> columnNameToTypeMap = Maps.newLinkedHashMap();
+ Set<String> ambiguousColumnNames = new HashSet<String>();
+ Map<String, Integer> fullColumnNameToTypeMap = Maps.newLinkedHashMap();
+ DatabaseMetaData dbmd = conn.getMetaData();
+ int unfoundColumnCount = 0;
+ // TODO: escape wildcard characters here because we don't want that
+ // behavior here
+ String escapedTableName = StringUtil.escapeLike(tableName);
+ String[] schemaAndTable = escapedTableName.split("\\.");
+ ResultSet rs = null;
+ try {
+ rs = dbmd.getColumns(null, (schemaAndTable.length == 1 ? ""
+ : schemaAndTable[0]),
+ (schemaAndTable.length == 1 ? escapedTableName
+ : schemaAndTable[1]), null);
+ while (rs.next()) {
+ String colName = rs.getString(QueryUtil.COLUMN_NAME_POSITION);
+ String colFam = rs.getString(QueryUtil.COLUMN_FAMILY_POSITION);
+
+ // use family qualifier, if available, otherwise, use column name
+ String fullColumn = (colFam==null?colName:String.format("%s.%s",colFam,colName));
+ String sqlTypeName = rs.getString(QueryUtil.DATA_TYPE_NAME_POSITION);
+
+ // allow for both bare and family qualified names.
+ if (columnNameToTypeMap.keySet().contains(colName)) {
+ ambiguousColumnNames.add(colName);
+ }
+ columnNameToTypeMap.put(
+ colName,
+ PDataType.fromSqlTypeName(sqlTypeName).getSqlType());
+ fullColumnNameToTypeMap.put(
+ fullColumn,
+ PDataType.fromSqlTypeName(sqlTypeName).getSqlType());
+ }
+ if (columnNameToTypeMap.isEmpty()) {
+ throw new IllegalArgumentException("Table " + tableName + " not found");
+ }
+ } finally {
+ if (rs != null) {
+ rs.close();
+ }
+ }
+ List<ColumnInfo> columnInfoList = Lists.newArrayList();
+ Set<String> unresolvedColumnNames = new TreeSet<String>();
+ if (columns == null) {
+ // use family qualified names by default, if no columns are specified.
+ for (Map.Entry<String, Integer> entry : fullColumnNameToTypeMap
+ .entrySet()) {
+ columnInfoList.add(new ColumnInfo(entry.getKey(), entry.getValue()));
+ }
+ } else {
+ // Leave "null" as indication to skip b/c it doesn't exist
+ for (int i = 0; i < columns.size(); i++) {
+ String columnName = columns.get(i).trim();
+ Integer sqlType = null;
+ if (fullColumnNameToTypeMap.containsKey(columnName)) {
+ sqlType = fullColumnNameToTypeMap.get(columnName);
+ } else if (columnNameToTypeMap.containsKey(columnName)) {
+ if (ambiguousColumnNames.contains(columnName)) {
+ unresolvedColumnNames.add(columnName);
+ }
+ // fall back to bare column name.
+ sqlType = columnNameToTypeMap.get(columnName);
+ }
+ if (unresolvedColumnNames.size()>0) {
+ StringBuilder exceptionMessage = new StringBuilder();
+ boolean first = true;
+ exceptionMessage.append("Unable to resolve these column names to a single column family:\n");
+ for (String col : unresolvedColumnNames) {
+ if (first) first = false;
+ else exceptionMessage.append(",");
+ exceptionMessage.append(col);
+ }
+ exceptionMessage.append("\nAvailable columns with column families:\n");
+ first = true;
+ for (String col : fullColumnNameToTypeMap.keySet()) {
+ if (first) first = false;
+ else exceptionMessage.append(",");
+ exceptionMessage.append(col);
+ }
+ throw new SQLException(exceptionMessage.toString());
+ }
+
+ if (sqlType == null) {
+ if (strict) {
+ throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode.COLUMN_NOT_FOUND)
+ .setColumnName(columnName)
+ .setTableName(tableName).build()
+ .buildException();
+ }
+ unfoundColumnCount++;
+ } else {
+ columnInfoList.add(new ColumnInfo(columnName, sqlType));
+ }
+ }
+ if (unfoundColumnCount == columns.size()) {
+ throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode.COLUMN_NOT_FOUND)
+ .setColumnName(
+ Arrays.toString(columns.toArray(new String[0])))
+ .setTableName(tableName).build().buildException();
+ }
+ }
+ return columnInfoList;
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/UpsertExecutor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/UpsertExecutor.java b/phoenix-core/src/main/java/org/apache/phoenix/util/UpsertExecutor.java
new file mode 100644
index 0000000..d9ce5f2
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/UpsertExecutor.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.util;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.phoenix.schema.types.PDataType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+
+/**
+ * Executes upsert statements on a provided {@code PreparedStatement} based on incoming
+ * {@code RECORDS}. An {@link UpsertListener} is notified each time the prepared statement
+ * is executed.
+ */
+public abstract class UpsertExecutor<RECORD, FIELD> implements Closeable {
+
+ /**
+ * A listener that is called for events based on incoming JSON data.
+ */
+ public interface UpsertListener<RECORD> {
+
+ /**
+ * Called when an upsert has been sucessfully completed. The given upsertCount is the total number of upserts
+ * completed on the caller up to this point.
+ *
+ * @param upsertCount total number of upserts that have been completed
+ */
+ void upsertDone(long upsertCount);
+
+
+ /**
+ * Called when executing a prepared statement has failed on a given record.
+ *
+ * @param record the JSON record that was being upserted when the error occurred
+ */
+ void errorOnRecord(RECORD record, Throwable throwable);
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(UpsertExecutor.class);
+
+ protected final Connection conn;
+ protected final List<ColumnInfo> columnInfos;
+ protected final List<PDataType> dataTypes;
+ protected final List<Function<FIELD, Object>> conversionFunctions;
+ protected final PreparedStatement preparedStatement;
+ protected final UpsertListener<RECORD> upsertListener;
+ protected long upsertCount = 0L;
+ protected boolean initFinished = false; // allow subclasses to finish initialization
+
+ private static PreparedStatement createStatement(Connection conn, String tableName,
+ List<ColumnInfo> columnInfoList) {
+ PreparedStatement preparedStatement;
+ try {
+ String upsertSql = QueryUtil.constructUpsertStatement(tableName, columnInfoList);
+ LOG.info("Upserting SQL data with {}", upsertSql);
+ preparedStatement = conn.prepareStatement(upsertSql);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ return preparedStatement;
+ }
+
+ /**
+ * Construct with the definition of incoming columns, and the statement upon which upsert
+ * statements are to be performed.
+ */
+ public UpsertExecutor(Connection conn, String tableName,
+ List<ColumnInfo> columnInfoList, UpsertListener<RECORD> upsertListener) {
+ this(conn, columnInfoList, createStatement(conn, tableName, columnInfoList), upsertListener);
+ }
+
+ /** Testing constructor. Do not use in prod. */
+ @VisibleForTesting
+ protected UpsertExecutor(Connection conn, List<ColumnInfo> columnInfoList,
+ PreparedStatement preparedStatement, UpsertListener<RECORD> upsertListener) {
+ this.conn = conn;
+ this.upsertListener = upsertListener;
+ this.columnInfos = columnInfoList;
+ this.preparedStatement = preparedStatement;
+ this.dataTypes = Lists.newArrayList();
+ this.conversionFunctions = Lists.newArrayList();
+ }
+
+ /**
+ * Awkward protocol allows subclass constructors to finish initializing context before
+ * proceeding to record processing.
+ */
+ protected void finishInit() {
+ for (ColumnInfo columnInfo : columnInfos) {
+ PDataType dataType = PDataType.fromTypeId(columnInfo.getSqlType());
+ dataTypes.add(dataType);
+ conversionFunctions.add(createConversionFunction(dataType));
+ }
+ this.initFinished = true;
+ }
+
+ /**
+ * Execute upserts for each JSON record contained in the given iterable, notifying this instance's
+ * {@code UpsertListener} for each completed upsert.
+ *
+ * @param records iterable of JSON records to be upserted
+ */
+ public void execute(Iterable<RECORD> records) {
+ if (!initFinished) {
+ finishInit();
+ }
+ for (RECORD record : records) {
+ execute(record);
+ }
+ }
+
+ /**
+ * Upsert a single record.
+ *
+ * @param record JSON record containing the data to be upserted
+ */
+ protected abstract void execute(RECORD record);
+
+ @Override
+ public void close() throws IOException {
+ try {
+ preparedStatement.close();
+ } catch (SQLException e) {
+ // An exception while closing the prepared statement is most likely a sign of a real problem, so we don't
+ // want to hide it with closeQuietly or something similar
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected abstract Function<FIELD, Object> createConversionFunction(PDataType dataType);
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/csv/CsvUpsertExecutor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/csv/CsvUpsertExecutor.java b/phoenix-core/src/main/java/org/apache/phoenix/util/csv/CsvUpsertExecutor.java
index e680f5c..4a3af21 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/csv/CsvUpsertExecutor.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/csv/CsvUpsertExecutor.java
@@ -17,132 +17,55 @@
*/
package org.apache.phoenix.util.csv;
-import java.io.Closeable;
-import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.util.List;
import java.util.Properties;
-
import javax.annotation.Nullable;
import org.apache.commons.csv.CSVRecord;
-import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PTimestamp;
import org.apache.phoenix.util.ColumnInfo;
import org.apache.phoenix.util.DateUtil;
-import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.UpsertExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
-import com.google.common.collect.Lists;
-/**
- * Executes upsert statements on a provided {@code PreparedStatement} based on incoming CSV records, notifying a
- * listener each time the prepared statement is executed.
- */
-public class CsvUpsertExecutor implements Closeable {
+/** {@link UpsertExecutor} over {@link CSVRecord}s. */
+public class CsvUpsertExecutor extends UpsertExecutor<CSVRecord, String> {
private static final Logger LOG = LoggerFactory.getLogger(CsvUpsertExecutor.class);
- private final String arrayElementSeparator;
- private final Connection conn;
- private final List<PDataType> dataTypes;
- private final List<Function<String,Object>> conversionFunctions;
- private final PreparedStatement preparedStatement;
- private final UpsertListener upsertListener;
- private long upsertCount = 0L;
-
- /**
- * A listener that is called for events based on incoming CSV data.
- */
- public static interface UpsertListener {
-
- /**
- * Called when an upsert has been sucessfully completed. The given upsertCount is the total number of upserts
- * completed on the caller up to this point.
- *
- * @param upsertCount total number of upserts that have been completed
- */
- void upsertDone(long upsertCount);
-
-
- /**
- * Called when executing a prepared statement has failed on a given record.
- *
- * @param csvRecord the CSV record that was being upserted when the error occurred
- */
- void errorOnRecord(CSVRecord csvRecord, Throwable throwable);
- }
-
-
- /**
- * Static constructor method for creating a CsvUpsertExecutor.
- *
- * @param conn Phoenix connection upon which upserts are to be performed
- * @param tableName name of the table in which upserts are to be performed
- * @param columnInfoList description of the columns to be upserted to, in the same order as in the CSV input
- * @param upsertListener listener that will be notified of upserts, can be null
- * @param arrayElementSeparator separator string to delimit string representations of arrays
- * @return the created CsvUpsertExecutor
- */
- public static CsvUpsertExecutor create(PhoenixConnection conn, String tableName, List<ColumnInfo> columnInfoList,
- UpsertListener upsertListener, String arrayElementSeparator) {
- PreparedStatement preparedStatement = null;
- try {
- String upsertSql = QueryUtil.constructUpsertStatement(tableName, columnInfoList);
- LOG.info("Upserting SQL data with {}", upsertSql);
- preparedStatement = conn.prepareStatement(upsertSql);
- } catch (SQLException e) {
- throw new RuntimeException(e);
- }
- return new CsvUpsertExecutor(conn, columnInfoList, preparedStatement, upsertListener,
- arrayElementSeparator);
- }
+ protected final String arrayElementSeparator;
- /**
- * Construct with the definition of incoming columns, and the statement upon which upsert statements
- * are to be performed.
- */
- CsvUpsertExecutor(Connection conn, List<ColumnInfo> columnInfoList, PreparedStatement preparedStatement,
- UpsertListener upsertListener, String arrayElementSeparator) {
- this.conn = conn;
- this.preparedStatement = preparedStatement;
- this.upsertListener = upsertListener;
+ /** Testing constructor. Do not use in prod. */
+ @VisibleForTesting
+ protected CsvUpsertExecutor(Connection conn, List<ColumnInfo> columnInfoList,
+ PreparedStatement stmt, UpsertListener<CSVRecord> upsertListener,
+ String arrayElementSeparator) {
+ super(conn, columnInfoList, stmt, upsertListener);
this.arrayElementSeparator = arrayElementSeparator;
- this.dataTypes = Lists.newArrayList();
- this.conversionFunctions = Lists.newArrayList();
- for (ColumnInfo columnInfo : columnInfoList) {
- PDataType dataType = PDataType.fromTypeId(columnInfo.getSqlType());
- dataTypes.add(dataType);
- conversionFunctions.add(createConversionFunction(dataType));
- }
+ finishInit();
}
- /**
- * Execute upserts for each CSV record contained in the given iterable, notifying this instance's
- * {@code UpsertListener} for each completed upsert.
- *
- * @param csvRecords iterable of CSV records to be upserted
- */
- public void execute(Iterable<CSVRecord> csvRecords) {
- for (CSVRecord csvRecord : csvRecords) {
- execute(csvRecord);
- }
+ public CsvUpsertExecutor(Connection conn, String tableName,
+ List<ColumnInfo> columnInfoList, UpsertListener<CSVRecord> upsertListener,
+ String arrayElementSeparator) {
+ super(conn, tableName, columnInfoList, upsertListener);
+ this.arrayElementSeparator = arrayElementSeparator;
+ finishInit();
}
- /**
- * Upsert a single record.
- *
- * @param csvRecord CSV record containing the data to be upserted
- */
- void execute(CSVRecord csvRecord) {
+ @Override
+ protected void execute(CSVRecord csvRecord) {
try {
if (csvRecord.size() < conversionFunctions.size()) {
String message = String.format("CSV record does not have enough values (has %d, but needs %d)",
@@ -170,17 +93,7 @@ public class CsvUpsertExecutor implements Closeable {
}
@Override
- public void close() throws IOException {
- try {
- preparedStatement.close();
- } catch (SQLException e) {
- // An exception while closing the prepared statement is most likely a sign of a real problem, so we don't
- // want to hide it with closeQuietly or something similar
- throw new RuntimeException(e);
- }
- }
-
- private Function<String, Object> createConversionFunction(PDataType dataType) {
+ protected Function<String, Object> createConversionFunction(PDataType dataType) {
if (dataType.isArrayType()) {
return new ArrayDatatypeConversionFunction(
new StringToArrayConverter(
@@ -201,7 +114,7 @@ public class CsvUpsertExecutor implements Closeable {
private final DateUtil.DateTimeParser dateTimeParser;
SimpleDatatypeConversionFunction(PDataType dataType, Connection conn) {
- Properties props = null;
+ Properties props;
try {
props = conn.getClientInfo();
} catch (SQLException e) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/json/JsonUpsertExecutor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/json/JsonUpsertExecutor.java b/phoenix-core/src/main/java/org/apache/phoenix/util/json/JsonUpsertExecutor.java
new file mode 100644
index 0000000..bbe0e30
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/json/JsonUpsertExecutor.java
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.util.json;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import javax.annotation.Nullable;
+
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PTimestamp;
+import org.apache.phoenix.util.ColumnInfo;
+import org.apache.phoenix.util.DateUtil;
+import org.apache.phoenix.util.UpsertExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.CaseFormat;
+import com.google.common.base.Function;
+
+/** {@link UpsertExecutor} over {@link Map} objects, as parsed from JSON. */
+public class JsonUpsertExecutor extends UpsertExecutor<Map<?, ?>, Object> {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(JsonUpsertExecutor.class);
+
+ /** Testing constructor. Do not use in prod. */
+ @VisibleForTesting
+ protected JsonUpsertExecutor(Connection conn, List<ColumnInfo> columnInfoList,
+ PreparedStatement stmt, UpsertListener<Map<?, ?>> upsertListener) {
+ super(conn, columnInfoList, stmt, upsertListener);
+ finishInit();
+ }
+
+ public JsonUpsertExecutor(Connection conn, String tableName, List<ColumnInfo> columnInfoList,
+ UpsertExecutor.UpsertListener<Map<?, ?>> upsertListener) {
+ super(conn, tableName, columnInfoList, upsertListener);
+ finishInit();
+ }
+
+ @Override
+ protected void execute(Map<?, ?> record) {
+ int fieldIndex = 0;
+ String colName = null;
+ try {
+ if (record.size() < conversionFunctions.size()) {
+ String message = String.format("JSON record does not have enough values (has %d, but needs %d)",
+ record.size(), conversionFunctions.size());
+ throw new IllegalArgumentException(message);
+ }
+ for (fieldIndex = 0; fieldIndex < conversionFunctions.size(); fieldIndex++) {
+ colName = CaseFormat.UPPER_UNDERSCORE.to(
+ CaseFormat.LOWER_UNDERSCORE, columnInfos.get(fieldIndex).getColumnName());
+ if (colName.contains(".")) {
+ StringBuilder sb = new StringBuilder();
+ String[] parts = colName.split("\\.");
+ // assume first part is the column family name; omita
+ for (int i = 1; i < parts.length; i++) {
+ sb.append(parts[i]);
+ if (i != parts.length - 1) {
+ sb.append(".");
+ }
+ }
+ colName = sb.toString();
+ }
+ if (colName.contains("\"")) {
+ colName = colName.replace("\"", "");
+ }
+ Object sqlValue = conversionFunctions.get(fieldIndex).apply(record.get(colName));
+ if (sqlValue != null) {
+ preparedStatement.setObject(fieldIndex + 1, sqlValue);
+ } else {
+ preparedStatement.setNull(fieldIndex + 1, dataTypes.get(fieldIndex).getSqlType());
+ }
+ }
+ preparedStatement.execute();
+ upsertListener.upsertDone(++upsertCount);
+ } catch (Exception e) {
+ if (LOG.isDebugEnabled()) {
+ // Even though this is an error we only log it with debug logging because we're notifying the
+ // listener, and it can do its own logging if needed
+ LOG.debug("Error on record " + record + ", fieldIndex " + fieldIndex + ", colName " + colName, e);
+ }
+ upsertListener.errorOnRecord(record, new Exception("fieldIndex: " + fieldIndex + ", colName " + colName, e));
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ preparedStatement.close();
+ } catch (SQLException e) {
+ // An exception while closing the prepared statement is most likely a sign of a real problem, so we don't
+ // want to hide it with closeQuietly or something similar
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected Function<Object, Object> createConversionFunction(PDataType dataType) {
+ if (dataType.isArrayType()) {
+ return new ArrayDatatypeConversionFunction(
+ new ObjectToArrayConverter(
+ conn,
+ PDataType.fromTypeId(dataType.getSqlType() - PDataType.ARRAY_TYPE_BASE)));
+ } else {
+ return new SimpleDatatypeConversionFunction(dataType, this.conn);
+ }
+ }
+
+ /**
+ * Performs typed conversion from String values to a given column value type.
+ */
+ static class SimpleDatatypeConversionFunction implements Function<Object, Object> {
+
+ private final PDataType dataType;
+ private final DateUtil.DateTimeParser dateTimeParser;
+
+ SimpleDatatypeConversionFunction(PDataType dataType, Connection conn) {
+ Properties props;
+ try {
+ props = conn.getClientInfo();
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ this.dataType = dataType;
+ if (dataType.isCoercibleTo(PTimestamp.INSTANCE)) {
+ // TODO: move to DateUtil
+ String dateFormat;
+ int dateSqlType = dataType.getResultSetSqlType();
+ if (dateSqlType == Types.DATE) {
+ dateFormat = props.getProperty(QueryServices.DATE_FORMAT_ATTRIB,
+ DateUtil.DEFAULT_DATE_FORMAT);
+ } else if (dateSqlType == Types.TIME) {
+ dateFormat = props.getProperty(QueryServices.TIME_FORMAT_ATTRIB,
+ DateUtil.DEFAULT_TIME_FORMAT);
+ } else {
+ dateFormat = props.getProperty(QueryServices.TIMESTAMP_FORMAT_ATTRIB,
+ DateUtil.DEFAULT_TIMESTAMP_FORMAT);
+ }
+ String timeZoneId = props.getProperty(QueryServices.DATE_FORMAT_TIMEZONE_ATTRIB,
+ QueryServicesOptions.DEFAULT_DATE_FORMAT_TIMEZONE);
+ this.dateTimeParser = DateUtil.getDateTimeParser(dateFormat, dataType, timeZoneId);
+ } else {
+ this.dateTimeParser = null;
+ }
+ }
+
+ @Nullable
+ @Override
+ public Object apply(@Nullable Object input) {
+ if (input == null) {
+ return null;
+ }
+ if (dateTimeParser != null && input instanceof String) {
+ final String s = (String) input;
+ long epochTime = dateTimeParser.parseDateTime(s);
+ byte[] byteValue = new byte[dataType.getByteSize()];
+ dataType.getCodec().encodeLong(epochTime, byteValue, 0);
+ return dataType.toObject(byteValue);
+ }
+ return dataType.toObject(input, dataType);
+ }
+ }
+
+ /**
+ * Converts string representations of arrays into Phoenix arrays of the correct type.
+ */
+ private static class ArrayDatatypeConversionFunction implements Function<Object, Object> {
+
+ private final ObjectToArrayConverter arrayConverter;
+
+ private ArrayDatatypeConversionFunction(ObjectToArrayConverter arrayConverter) {
+ this.arrayConverter = arrayConverter;
+ }
+
+ @Nullable
+ @Override
+ public Object apply(@Nullable Object input) {
+ try {
+ return arrayConverter.toArray(input);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/json/ObjectToArrayConverter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/json/ObjectToArrayConverter.java b/phoenix-core/src/main/java/org/apache/phoenix/util/json/ObjectToArrayConverter.java
new file mode 100644
index 0000000..16bef15
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/json/ObjectToArrayConverter.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.util.json;
+
+import java.sql.Array;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.phoenix.schema.types.PDataType;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+
+/**
+ * Converts Objects (presumably lists) into Phoenix arrays.
+ */
+class ObjectToArrayConverter {
+
+ private final Connection conn;
+ private final PDataType elementDataType;
+ private final JsonUpsertExecutor.SimpleDatatypeConversionFunction elementConvertFunction;
+
+ /**
+ * Instantiate with the array value separator and data type.
+ *
+ * @param conn Phoenix connection to target database
+ * @param elementDataType datatype of the elements of arrays to be created
+ */
+ public ObjectToArrayConverter(Connection conn, PDataType elementDataType) {
+ this.conn = conn;
+ this.elementDataType = elementDataType;
+ this.elementConvertFunction =
+ new JsonUpsertExecutor.SimpleDatatypeConversionFunction(elementDataType, this.conn);
+ }
+
+ /**
+ * Convert an input delimited string into a phoenix array of the configured type.
+ *
+ * @param input string containing delimited array values
+ * @return the array containing the values represented in the input string
+ */
+ public Array toArray(Object input) throws SQLException {
+ if (input == null) {
+ return conn.createArrayOf(elementDataType.getSqlTypeName(), new Object[0]);
+ }
+ List<?> list = (List<?>) input;
+ if (list.isEmpty()) {
+ return conn.createArrayOf(elementDataType.getSqlTypeName(), new Object[0]);
+ }
+ return conn.createArrayOf(elementDataType.getSqlTypeName(),
+ Lists.newArrayList(Iterables.transform(list, elementConvertFunction)).toArray());
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/BulkLoadToolTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/BulkLoadToolTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/BulkLoadToolTest.java
new file mode 100644
index 0000000..95e9b43
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/BulkLoadToolTest.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.mapreduce;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.apache.commons.cli.CommandLine;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.junit.Assert.assertEquals;
+
+@RunWith(Parameterized.class)
+public class BulkLoadToolTest {
+
+ @Parameterized.Parameters
+ public static Collection<Object[]> params() {
+ return Arrays.asList(new Object[][]{
+ { new CsvBulkLoadTool() },
+ { new JsonBulkLoadTool() },
+ });
+ }
+
+ @Parameterized.Parameter(value = 0)
+ public AbstractBulkLoadTool bulkLoadTool;
+
+ @Test
+ public void testParseOptions() {
+ CommandLine cmdLine = bulkLoadTool.parseOptions(new String[] { "--input", "/input",
+ "--table", "mytable" });
+
+ assertEquals("mytable", cmdLine.getOptionValue(CsvBulkLoadTool.TABLE_NAME_OPT.getOpt()));
+ assertEquals("/input", cmdLine.getOptionValue(CsvBulkLoadTool.INPUT_PATH_OPT.getOpt()));
+ }
+
+ @Test(expected=IllegalStateException.class)
+ public void testParseOptions_ExtraArguments() {
+ bulkLoadTool.parseOptions(new String[] { "--input", "/input",
+ "--table", "mytable", "these", "shouldnt", "be", "here" });
+ }
+
+ @Test(expected=IllegalStateException.class)
+ public void testParseOptions_NoInput() {
+ bulkLoadTool.parseOptions(new String[] { "--table", "mytable" });
+ }
+
+ @Test(expected=IllegalStateException.class)
+ public void testParseOptions_NoTable() {
+ bulkLoadTool.parseOptions(new String[] { "--input", "/input" });
+ }
+
+ @Test
+ public void testGetQualifiedTableName() {
+ assertEquals("MYSCHEMA.MYTABLE", CsvBulkLoadTool.getQualifiedTableName("mySchema", "myTable"));
+ }
+
+ @Test
+ public void testGetQualifiedTableName_NullSchema() {
+ assertEquals("MYTABLE", CsvBulkLoadTool.getQualifiedTableName(null, "myTable"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java
index f52a837..3c6271a 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java
@@ -17,9 +17,6 @@
*/
package org.apache.phoenix.mapreduce;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
@@ -29,11 +26,10 @@ import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
-import org.apache.phoenix.schema.types.PInteger;
-import org.apache.phoenix.util.ColumnInfo;
import org.junit.Test;
-import com.google.common.collect.ImmutableList;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
public class CsvBulkImportUtilTest {
@@ -41,16 +37,11 @@ public class CsvBulkImportUtilTest {
public void testInitCsvImportJob() throws IOException {
Configuration conf = new Configuration();
- String tableName = "SCHEMANAME.TABLENAME";
char delimiter = '\001';
char quote = '\002';
char escape = '!';
- List<ColumnInfo> columnInfoList = ImmutableList.of(
- new ColumnInfo("MYCOL", PInteger.INSTANCE.getSqlType()));
-
- CsvBulkImportUtil.initCsvImportJob(
- conf, tableName, delimiter, quote, escape, null, columnInfoList, true);
+ CsvBulkImportUtil.initCsvImportJob(conf, delimiter, quote, escape, null);
// Serialize and deserialize the config to ensure that there aren't any issues
// with non-printable characters as delimiters
@@ -61,7 +52,6 @@ public class CsvBulkImportUtilTest {
Configuration deserialized = new Configuration();
deserialized.addResource(new FileInputStream(tempFile));
- assertEquals(tableName, deserialized.get(CsvToKeyValueMapper.TABLE_NAME_CONFKEY));
assertEquals(Character.valueOf('\001'),
CsvBulkImportUtil.getCharacter(deserialized, CsvToKeyValueMapper.FIELD_DELIMITER_CONFKEY));
assertEquals(Character.valueOf('\002'),
@@ -69,8 +59,6 @@ public class CsvBulkImportUtilTest {
assertEquals(Character.valueOf('!'),
CsvBulkImportUtil.getCharacter(deserialized, CsvToKeyValueMapper.ESCAPE_CHAR_CONFKEY));
assertNull(deserialized.get(CsvToKeyValueMapper.ARRAY_DELIMITER_CONFKEY));
- assertEquals(columnInfoList, CsvToKeyValueMapper.buildColumnInfoList(deserialized));
- assertEquals(true, deserialized.getBoolean(CsvToKeyValueMapper.IGNORE_INVALID_ROW_CONFKEY, false));
tempFile.delete();
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolTest.java
deleted file mode 100644
index 33bb976..0000000
--- a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolTest.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.phoenix.mapreduce;
-
-import org.apache.commons.cli.CommandLine;
-import org.junit.Before;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
-public class CsvBulkLoadToolTest {
-
- private CsvBulkLoadTool bulkLoadTool;
-
- @Before
- public void setUp() {
- bulkLoadTool = new CsvBulkLoadTool();
- }
-
- @Test
- public void testParseOptions() {
- CommandLine cmdLine = bulkLoadTool.parseOptions(new String[] { "--input", "/input",
- "--table", "mytable" });
-
- assertEquals("mytable", cmdLine.getOptionValue(CsvBulkLoadTool.TABLE_NAME_OPT.getOpt()));
- assertEquals("/input", cmdLine.getOptionValue(CsvBulkLoadTool.INPUT_PATH_OPT.getOpt()));
- }
-
- @Test(expected=IllegalStateException.class)
- public void testParseOptions_ExtraArguments() {
- bulkLoadTool.parseOptions(new String[] { "--input", "/input",
- "--table", "mytable", "these", "shouldnt", "be", "here" });
- }
-
- @Test(expected=IllegalStateException.class)
- public void testParseOptions_NoInput() {
- bulkLoadTool.parseOptions(new String[] { "--table", "mytable" });
- }
-
- @Test(expected=IllegalStateException.class)
- public void testParseOptions_NoTable() {
- bulkLoadTool.parseOptions(new String[] { "--input", "/input" });
- }
-
- @Test
- public void testGetQualifiedTableName() {
- assertEquals("MYSCHEMA.MYTABLE", CsvBulkLoadTool.getQualifiedTableName("mySchema", "myTable"));
- }
-
- @Test
- public void testGetQualifiedTableName_NullSchema() {
- assertEquals("MYTABLE", CsvBulkLoadTool.getQualifiedTableName(null, "myTable"));
- }
-}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java
index dc6f497..fe4e068 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java
@@ -17,25 +17,13 @@
*/
package org.apache.phoenix.mapreduce;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
import java.io.IOException;
-import java.util.List;
import org.apache.commons.csv.CSVRecord;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
-import org.apache.phoenix.schema.types.PInteger;
-import org.apache.phoenix.schema.types.PIntegerArray;
-import org.apache.phoenix.schema.types.PUnsignedInt;
-import org.apache.phoenix.util.ColumnInfo;
import org.junit.Test;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
public class CsvToKeyValueMapperTest {
@@ -62,72 +50,4 @@ public class CsvToKeyValueMapperTest {
assertTrue(parsed.isConsistent());
assertEquals(1, parsed.getRecordNumber());
}
-
-
- @Test
- public void testBuildColumnInfoList() {
- List<ColumnInfo> columnInfoList = ImmutableList.of(
- new ColumnInfo("idCol", PInteger.INSTANCE.getSqlType()),
- new ColumnInfo("unsignedIntCol", PUnsignedInt.INSTANCE.getSqlType()),
- new ColumnInfo("stringArrayCol", PIntegerArray.INSTANCE.getSqlType()));
-
- Configuration conf = new Configuration();
- CsvToKeyValueMapper.configureColumnInfoList(conf, columnInfoList);
- List<ColumnInfo> fromConfig = CsvToKeyValueMapper.buildColumnInfoList(conf);
-
- assertEquals(columnInfoList, fromConfig);
- }
-
- @Test
- public void testBuildColumnInfoList_ContainingNulls() {
- // A null value in the column info list means "skip that column in the input"
- List<ColumnInfo> columnInfoListWithNull = Lists.newArrayList(
- new ColumnInfo("idCol", PInteger.INSTANCE.getSqlType()),
- null,
- new ColumnInfo("unsignedIntCol", PUnsignedInt.INSTANCE.getSqlType()),
- new ColumnInfo("stringArrayCol", PIntegerArray.INSTANCE.getSqlType()));
-
- Configuration conf = new Configuration();
- CsvToKeyValueMapper.configureColumnInfoList(conf, columnInfoListWithNull);
- List<ColumnInfo> fromConfig = CsvToKeyValueMapper.buildColumnInfoList(conf);
-
- assertEquals(columnInfoListWithNull, fromConfig);
- }
-
- @Test
- public void testLoadPreUpdateProcessor() {
- Configuration conf = new Configuration();
- conf.setClass(PhoenixConfigurationUtil.UPSERT_HOOK_CLASS_CONFKEY, MockUpsertProcessor.class,
- ImportPreUpsertKeyValueProcessor.class);
-
- ImportPreUpsertKeyValueProcessor processor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
- assertEquals(MockUpsertProcessor.class, processor.getClass());
- }
-
- @Test
- public void testLoadPreUpdateProcessor_NotConfigured() {
-
- Configuration conf = new Configuration();
- ImportPreUpsertKeyValueProcessor processor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
-
- assertEquals(CsvToKeyValueMapper.DefaultImportPreUpsertKeyValueProcessor.class,
- processor.getClass());
- }
-
- @Test(expected=IllegalStateException.class)
- public void testLoadPreUpdateProcessor_ClassNotFound() {
- Configuration conf = new Configuration();
- conf.set(PhoenixConfigurationUtil.UPSERT_HOOK_CLASS_CONFKEY, "MyUndefinedClass");
-
- PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
- }
-
-
- static class MockUpsertProcessor implements ImportPreUpsertKeyValueProcessor {
-
- @Override
- public List<KeyValue> preUpsert(byte[] rowKey, List<KeyValue> keyValues) {
- throw new UnsupportedOperationException("Not yet implemented");
- }
- }
}