Removing old gora-cassandra module
Project: http://git-wip-us.apache.org/repos/asf/gora/repo Commit: http://git-wip-us.apache.org/repos/asf/gora/commit/cbed41d0 Tree: http://git-wip-us.apache.org/repos/asf/gora/tree/cbed41d0 Diff: http://git-wip-us.apache.org/repos/asf/gora/diff/cbed41d0 Branch: refs/heads/master Commit: cbed41d00ab55776b8f7647ffe00818ff736ddba Parents: 3d4e1d6 Author: Lewis John McGibbney <lewis.mcgibb...@gmail.com> Authored: Wed Aug 23 14:05:38 2017 -0700 Committer: Lewis John McGibbney <lewis.mcgibb...@gmail.com> Committed: Wed Aug 23 14:05:38 2017 -0700 ---------------------------------------------------------------------- gora-cassandra/conf/.gitignore | 15 - gora-cassandra/pom.xml | 192 ------ gora-cassandra/src/examples/java/.gitignore | 15 - .../org/apache/gora/cassandra/package-info.java | 20 - .../gora/cassandra/query/CassandraColumn.java | 97 --- .../gora/cassandra/query/CassandraQuery.java | 107 --- .../gora/cassandra/query/CassandraResult.java | 194 ------ .../cassandra/query/CassandraResultSet.java | 54 -- .../gora/cassandra/query/CassandraRow.java | 58 -- .../cassandra/query/CassandraSubColumn.java | 111 --- .../cassandra/query/CassandraSuperColumn.java | 197 ------ .../gora/cassandra/query/package-info.java | 21 - .../serializers/AvroSerializerUtil.java | 94 --- .../serializers/CharSequenceSerializer.java | 64 -- .../serializers/GoraSerializerTypeInferer.java | 230 ------- .../cassandra/serializers/ListSerializer.java | 193 ------ .../cassandra/serializers/MapSerializer.java | 223 ------- .../serializers/SpecificFixedSerializer.java | 119 ---- .../gora/cassandra/serializers/TypeUtils.java | 232 ------- .../cassandra/serializers/package-info.java | 20 - .../gora/cassandra/store/CassandraClient.java | 658 ------------------ .../gora/cassandra/store/CassandraMapping.java | 329 --------- .../store/CassandraMappingManager.java | 158 ----- .../gora/cassandra/store/CassandraStore.java | 668 ------------------- .../gora/cassandra/store/HectorUtils.java | 117 ---- .../gora/cassandra/store/package-info.java | 20 - gora-cassandra/src/test/conf/cassandra.yaml | 418 ------------ .../src/test/conf/gora-cassandra-mapping.xml | 95 --- gora-cassandra/src/test/conf/gora.properties | 28 - .../src/test/conf/log4j-server.properties | 47 -- gora-cassandra/src/test/java/.gitignore | 15 - .../gora/cassandra/GoraCassandraTestDriver.java | 158 ----- .../org/apache/gora/cassandra/package-info.java | 21 - .../cassandra/store/TestCassandraStore.java | 92 --- .../gora/cassandra/store/package-info.java | 21 - 35 files changed, 5101 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/conf/.gitignore ---------------------------------------------------------------------- diff --git a/gora-cassandra/conf/.gitignore b/gora-cassandra/conf/.gitignore deleted file mode 100644 index 09697dc..0000000 --- a/gora-cassandra/conf/.gitignore +++ /dev/null @@ -1,15 +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. - http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/pom.xml ---------------------------------------------------------------------- diff --git a/gora-cassandra/pom.xml b/gora-cassandra/pom.xml deleted file mode 100644 index 0d3d13c..0000000 --- a/gora-cassandra/pom.xml +++ /dev/null @@ -1,192 +0,0 @@ -<?xml version="1.0" encoding="UTF-8"?> -<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> - <!-- - 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. - --> - <modelVersion>4.0.0</modelVersion> - - <parent> - <groupId>org.apache.gora</groupId> - <artifactId>gora</artifactId> - <version>0.8-SNAPSHOT</version> - <relativePath>../</relativePath> - </parent> - <artifactId>gora-cassandra</artifactId> - <packaging>bundle</packaging> - - <name>Apache Gora :: Cassandra</name> - <url>http://gora.apache.org</url> - <description>The Apache Gora open source framework provides an in-memory data model and - persistence for big data. Gora supports persisting to column stores, key value stores, - document stores and RDBMSs, and analyzing the data with extensive Apache Hadoop MapReduce - support.</description> - <inceptionYear>2010</inceptionYear> - <organization> - <name>The Apache Software Foundation</name> - <url>http://www.apache.org/</url> - </organization> - <issueManagement> - <system>JIRA</system> - <url>https://issues.apache.org/jira/browse/GORA</url> - </issueManagement> - <ciManagement> - <system>Jenkins</system> - <url>https://builds.apache.org/job/Gora-trunk/</url> - </ciManagement> - - <properties> - <osgi.import>*</osgi.import> - <osgi.export>org.apache.gora.cassandra*;version="${project.version}";-noimport:=true</osgi.export> - </properties> - - <build> - <directory>target</directory> - <outputDirectory>target/classes</outputDirectory> - <finalName>${project.artifactId}-${project.version}</finalName> - <testOutputDirectory>target/test-classes</testOutputDirectory> - <testSourceDirectory>src/test/java</testSourceDirectory> - <sourceDirectory>src/main/java</sourceDirectory> - <testResources> - <testResource> - <directory>${project.basedir}/src/test/conf</directory> - <includes> - <include>**/*</include> - </includes> - <!--targetPath>${project.basedir}/target/classes/</targetPath--> - </testResource> - </testResources> - <plugins> - <plugin> - <groupId>org.codehaus.mojo</groupId> - <artifactId>build-helper-maven-plugin</artifactId> - <version>${build-helper-maven-plugin.version}</version> - <executions> - <execution> - <phase>generate-sources</phase> - <goals> - <goal>add-source</goal> - </goals> - <configuration> - <sources> - <source>src/examples/java</source> - </sources> - </configuration> - </execution> - </executions> - </plugin> - </plugins> - </build> - - <dependencies> - - <!-- OSX JDK 7 --> - <!-- should be removed once fixed in Cassandra --> - <dependency> - <groupId>org.xerial.snappy</groupId> - <artifactId>snappy-java</artifactId> - <version>1.0.5-M3</version> - <scope>test</scope> - </dependency> - - <!-- Gora Internal Dependencies --> - <dependency> - <groupId>org.apache.gora</groupId> - <artifactId>gora-core</artifactId> - </dependency> - - <dependency> - <groupId>org.apache.gora</groupId> - <artifactId>gora-core</artifactId> - <type>test-jar</type> - <scope>test</scope> - </dependency> - - <!-- Cassandra Dependencies --> - <dependency> - <groupId>org.apache.cassandra</groupId> - <artifactId>cassandra-all</artifactId> - <scope>test</scope> - <exclusions> - <exclusion> - <groupId>org.apache.cassandra.deps</groupId> - <artifactId>avro</artifactId> - </exclusion> - </exclusions> - </dependency> - - <!-- dependency> - <groupId>org.apache.cassandra</groupId> - <artifactId>cassandra-thrift</artifactId> - </dependency--> - - <dependency> - <groupId>org.hectorclient</groupId> - <artifactId>hector-core</artifactId> - <exclusions> - <exclusion> - <groupId>org.apache.cassandra</groupId> - <artifactId>cassandra-all</artifactId> - </exclusion> - </exclusions> - </dependency> - - <!-- Misc Dependencies --> - <dependency> - <groupId>com.google.guava</groupId> - <artifactId>guava</artifactId> - </dependency> - - <dependency> - <groupId>org.jdom</groupId> - <artifactId>jdom</artifactId> - </dependency> - - - <dependency> - <groupId>org.apache.avro</groupId> - <artifactId>avro</artifactId> - </dependency> - - <!-- Logging Dependencies --> - <dependency> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-log4j12</artifactId> - </dependency> - <dependency> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - <exclusions> - <exclusion> - <groupId>javax.jms</groupId> - <artifactId>jms</artifactId> - </exclusion> - </exclusions> - </dependency> - - <!-- Testing Dependencies --> - <dependency> - <groupId>junit</groupId> - <artifactId>junit</artifactId> - </dependency> - - <dependency> - <groupId>org.apache.hadoop</groupId> - <artifactId>hadoop-minicluster</artifactId> - </dependency> - - </dependencies> - -</project> http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/examples/java/.gitignore ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/examples/java/.gitignore b/gora-cassandra/src/examples/java/.gitignore deleted file mode 100644 index 09697dc..0000000 --- a/gora-cassandra/src/examples/java/.gitignore +++ /dev/null @@ -1,15 +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. - http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/package-info.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/package-info.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/package-info.java deleted file mode 100644 index 5247ecc..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/package-info.java +++ /dev/null @@ -1,20 +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. - */ -/** - * This package contains Casandra datastore related all classes. - */ -package org.apache.gora.cassandra; \ No newline at end of file http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraColumn.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraColumn.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraColumn.java deleted file mode 100644 index 3cffdbf..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraColumn.java +++ /dev/null @@ -1,97 +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.gora.cassandra.query; - -import java.io.IOException; -import java.nio.ByteBuffer; - -import me.prettyprint.hector.api.Serializer; - -import org.apache.avro.Schema; -import org.apache.avro.Schema.Field; -import org.apache.avro.Schema.Type; -import org.apache.gora.cassandra.serializers.AvroSerializerUtil; -import org.apache.gora.cassandra.serializers.GoraSerializerTypeInferer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Represents a unit of data: a key value pair tagged by a family name - */ -public abstract class CassandraColumn { - public static final Logger LOG = LoggerFactory.getLogger(CassandraColumn.class); - - public static final int SUB = 0; - public static final int SUPER = 1; - - private String family; - private int type; - private Field field; - private int unionType; - - public void setUnionType(int pUnionType){ - this.unionType = pUnionType; - } - - public int getUnionType(){ - return unionType; - } - - public String getFamily() { - return family; - } - public void setFamily(String family) { - this.family = family; - } - public int getType() { - return type; - } - public void setType(int type) { - this.type = type; - } - public void setField(Field field) { - this.field = field; - } - - protected Field getField() { - return this.field; - } - - public abstract ByteBuffer getName(); - public abstract Object getValue(); - - protected Object fromByteBuffer(Schema schema, ByteBuffer byteBuffer) { - Object value = null; - Serializer<?> serializer = GoraSerializerTypeInferer.getSerializer(schema); - if (serializer == null) { - LOG.warn("Schema: " + schema.getName() + " is not supported. No serializer " - + "could be found. Please report this to d...@gora.apache.org"); - } else { - value = serializer.fromByteBuffer(byteBuffer); - if (schema.getType().equals(Type.RECORD) || schema.getType().equals(Type.MAP) ){ - try { - value = AvroSerializerUtil.deserializer(value, schema); - } catch (IOException e) { - LOG.warn(field.name() + " named field could not be deserialized."); - } - } - } - return value; - } -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java deleted file mode 100644 index 73e7c54..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java +++ /dev/null @@ -1,107 +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.gora.cassandra.query; - -import java.util.List; -import java.util.Map; - -import org.apache.gora.persistency.impl.PersistentBase; -import org.apache.gora.query.Query; -import org.apache.gora.query.impl.QueryBase; -import org.apache.gora.store.DataStore; - -/** - * Cassandra specific implementation of the {@link Query} interface. - */ -public class CassandraQuery<K, T extends PersistentBase> extends QueryBase<K, T> { - - /** - * Cassandra specific implementation of the {@link Query} interface. - */ - private Query<K, T> query; - - /** - * Maps Avro fields to Cassandra columns. - */ - private Map<String, List<String>> familyMap; - - /** - * Constructor for the query - */ - public CassandraQuery() { - super(null); - } - - /** - * Constructor for the query - * - * @param dataStore Data store used - * - */ - public CassandraQuery(DataStore<K, T> dataStore) { - super(dataStore); - } - - /** - * Setter of familyMap. - */ - public void setFamilyMap(Map<String, List<String>> familyMap) { - this.familyMap = familyMap; - } - - /** - * Getter of familyMap. - */ - public Map<String, List<String>> getFamilyMap() { - return familyMap; - } - - /** - * @param family the family name - * @return an array of the query column names belonging to the family - */ - public String[] getColumns(String family) { - - List<String> columnList = familyMap.get(family); - String[] columns = new String[columnList.size()]; - for (int i = 0; i < columns.length; ++i) { - columns[i] = columnList.get(i); - } - return columns; - } - - /** - * - * @return get {@link org.apache.gora.cassandra.query.CassandraQuery} - */ - public Query<K, T> getQuery() { - return query; - } - - /** - * - * @param query set a {@link org.apache.gora.cassandra.query.CassandraQuery} - */ - public void setQuery(Query<K, T> query) { - this.query = query; - } - - - -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java deleted file mode 100644 index a89c97c..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java +++ /dev/null @@ -1,194 +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.gora.cassandra.query; - -import java.io.IOException; -import java.util.List; -import java.util.Map; - -import me.prettyprint.cassandra.serializers.StringSerializer; - -import org.apache.avro.Schema; -import org.apache.avro.Schema.Field; -import org.apache.avro.Schema.Type; -import org.apache.gora.cassandra.store.CassandraStore; -import org.apache.gora.persistency.impl.PersistentBase; -import org.apache.gora.query.Query; -import org.apache.gora.query.impl.ResultBase; -import org.apache.gora.store.DataStore; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * CassandraResult specific implementation of the {@link org.apache.gora.query.Result} - * interface. - */ -public class CassandraResult<K, T extends PersistentBase> extends ResultBase<K, T> { - public static final Logger LOG = LoggerFactory.getLogger(CassandraResult.class); - - private int rowNumber; - - /** - * Result set containing query results - */ - private CassandraResultSet<K> cassandraResultSet; - - /** - * Maps Cassandra columns to Avro fields. - */ - private Map<String, String> reverseMap; - - - /** - * Constructor for the result set - * - * @param dataStore Data store used - * @param query Query used - */ - public CassandraResult(DataStore<K, T> dataStore, Query<K, T> query) { - super(dataStore, query); - } - - /** - * Gets the next item - */ - @Override - protected boolean nextInner() throws IOException { - if (this.rowNumber < this.cassandraResultSet.size()) { - updatePersistent(); - } - ++this.rowNumber; - return (this.rowNumber <= this.cassandraResultSet.size()); - } - - /** - * Gets the column containing the type of the union type element stored. - * TODO: This might seem too much of a overhead if we consider that N rows have M columns, - * this might have to be reviewed to get the specific column in O(1) - * @param pFieldName - * @param pCassandraRow - * @return - */ - private CassandraColumn getUnionTypeColumn(String pFieldName, Object[] pCassandraRow){ - - for (Object currentPCassandraRow : pCassandraRow) { - CassandraColumn cColumn = (CassandraColumn) currentPCassandraRow; - String columnName = StringSerializer.get().fromByteBuffer(cColumn.getName().duplicate()); - if (pFieldName.equals(columnName)) - return cColumn; - } - return null; - } - - - /** - * Load key/value pair from Cassandra row to Avro record. - * @throws IOException - */ - private void updatePersistent() throws IOException { - CassandraRow<K> cassandraRow = this.cassandraResultSet.get(this.rowNumber); - - // load key - this.key = cassandraRow.getKey(); - - // load value - Schema schema = this.persistent.getSchema(); - List<Field> fields = schema.getFields(); - - for (CassandraColumn cassandraColumn: cassandraRow) { - // get field name - String family = cassandraColumn.getFamily(); - - String fieldName = this.reverseMap.get(family + ":" + StringSerializer.get().fromByteBuffer(cassandraColumn.getName().duplicate())); - - if (fieldName != null) { - // get field - if (!fieldName.contains(CassandraStore.UNION_COL_SUFIX)) { - - int pos = this.persistent.getSchema().getField(fieldName).pos(); - Field field = fields.get(pos); - Type fieldType = field.schema().getType(); - if (fieldType.equals(Type.UNION)) { - //getting UNION stored type - CassandraColumn cc = getUnionTypeColumn(fieldName - + CassandraStore.UNION_COL_SUFIX, cassandraRow.toArray()); - //creating temporary UNION Field - Field unionField = new Field(fieldName - + CassandraStore.UNION_COL_SUFIX, Schema.create(Type.INT), - null, null); - // get value of UNION stored type - cc.setField(unionField); - Object val = cc.getValue(); - cassandraColumn.setUnionType(Integer.parseInt(val.toString())); - } - - // get value - cassandraColumn.setField(field); - Object value = cassandraColumn.getValue(); - - this.persistent.put(pos, value); - // this field does not need to be written back to the store - this.persistent.clearDirty(pos); - } - } else - LOG.debug("FieldName was null while iterating CassandraRow and using Avro Union type"); - } - - } - - //TODO Should we remove this method? - @SuppressWarnings("unused") - private int getNonNullTypePos(List<Schema> pTypes){ - int iCnt = 0; - for (Schema sch : pTypes) - if (!sch.getName().equals("null")) - return iCnt; - else - iCnt++; - return CassandraStore.DEFAULT_UNION_SCHEMA; - } - - @Override - public void close() throws IOException { - // TODO Auto-generated method stub - - } - - /** - * Gets the items reading progress - */ - @Override - public float getProgress() throws IOException { - return (((float) this.rowNumber) / this.cassandraResultSet.size()); - } - - /** - * Set the Result set containing query results - * - * @param cassandraResultSet - */ - public void setResultSet(CassandraResultSet<K> cassandraResultSet) { - this.cassandraResultSet = cassandraResultSet; - } - - public void setReverseMap(Map<String, String> reverseMap) { - this.reverseMap = reverseMap; - } - -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResultSet.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResultSet.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResultSet.java deleted file mode 100644 index c994860..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResultSet.java +++ /dev/null @@ -1,54 +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.gora.cassandra.query; - -import java.util.ArrayList; -import java.util.HashMap; - -/** - * List data structure to keep the order coming from the Cassandra selects. - */ -public class CassandraResultSet<K> extends ArrayList<CassandraRow<K>> { - - /** - * - */ - private static final long serialVersionUID = -7620939600192859652L; - - /** - * Maps keys to indices in the list. - */ - private HashMap<K, Integer> indexMap = new HashMap<>(); - - public CassandraRow<K> getRow(K key) { - Integer integer = this.indexMap.get(key); - if (integer == null) { - return null; - } - - return this.get(integer); - } - - public void putRow(K key, CassandraRow<K> cassandraRow) { - this.add(cassandraRow); - this.indexMap.put(key, this.size()-1); - } - - -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraRow.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraRow.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraRow.java deleted file mode 100644 index 685d8e4..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraRow.java +++ /dev/null @@ -1,58 +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.gora.cassandra.query; - -import java.util.ArrayList; - -import me.prettyprint.cassandra.serializers.StringSerializer; - - -/** - * List of key value pairs representing a row, tagged by a key. - */ -public class CassandraRow<K> extends ArrayList<CassandraColumn> { - - /** - * - */ - private static final long serialVersionUID = -7620939600192859652L; - private K key; - - public K getKey() { - return this.key; - } - - public void setKey(K key) { - this.key = key; - } - - /** - * Gets a specific CassandraColumn within a row using its name - * @param pCassandraColumnName - * @return CassandraColumn - */ - public CassandraColumn getCassandraColumn(String pCassandraColumnName){ - for (CassandraColumn cColumn: this) - if ( pCassandraColumnName.equals(StringSerializer.get().fromByteBuffer(cColumn.getName())) ) - return cColumn; - - return null; - } - -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSubColumn.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSubColumn.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSubColumn.java deleted file mode 100644 index efd0c20..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSubColumn.java +++ /dev/null @@ -1,111 +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.gora.cassandra.query; - -import java.nio.ByteBuffer; -import java.util.List; - -import me.prettyprint.hector.api.beans.HColumn; - -import org.apache.avro.Schema; -import org.apache.avro.Schema.Field; -import org.apache.avro.Schema.Type; -import org.apache.gora.cassandra.serializers.ListSerializer; -import org.apache.gora.cassandra.store.CassandraStore; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public class CassandraSubColumn extends CassandraColumn { - public static final Logger LOG = LoggerFactory.getLogger(CassandraSubColumn.class); - - /** - * Key-value pair containing the raw data. - */ - private HColumn<ByteBuffer, ByteBuffer> hColumn; - - public ByteBuffer getName() { - return hColumn.getName(); - } - - private Object getFieldValue(Type type, Schema fieldSchema, ByteBuffer byteBuffer){ - Object value = null; - if (type.equals(Type.ARRAY)) { - ListSerializer<?> serializer = ListSerializer.get(fieldSchema.getElementType()); - if (serializer != null) { - List<?> genericArray = serializer.fromByteBuffer(byteBuffer); - value = genericArray; - } else { - LOG.warn("Field detected as type Array, however no serializer could be found!"); - } - } else if (type.equals(Type.MAP)) { -// MapSerializer<?> serializer = MapSerializer.get(fieldSchema.getValueType()); -// Map<?, ?> map = serializer.fromByteBuffer(byteBuffer); -// value = map; - value = fromByteBuffer(fieldSchema, byteBuffer); - } else if (type.equals(Type.RECORD)){ - value = fromByteBuffer(fieldSchema, byteBuffer); - } else if (type.equals(Type.UNION)){ - // the selected union schema is obtained - Schema unionFieldSchema = getUnionSchema(super.getUnionType(), fieldSchema); - Type unionFieldType = unionFieldSchema.getType(); - // we use the selected union schema to deserialize our actual value - //value = fromByteBuffer(unionFieldSchema, byteBuffer); - value = getFieldValue(unionFieldType, unionFieldSchema, byteBuffer); - } else { - value = fromByteBuffer(fieldSchema, byteBuffer); - } - return value; - } - - /** - * Deserialize a String into an typed Object, according to the field schema. - * @see org.apache.gora.cassandra.query.CassandraColumn#getValue() - */ - public Object getValue() { - Field field = getField(); - Schema fieldSchema = field.schema(); - Type type = fieldSchema.getType(); - ByteBuffer byteBuffer = hColumn.getValue(); - if (byteBuffer == null) { - return null; - } - - Object value = getFieldValue(type, fieldSchema, byteBuffer); - return value; - } - - /** - * Gets the specific schema for a union data type - * @param pSchemaPos - * @param pSchema - * @return - */ - private Schema getUnionSchema (int pSchemaPos, Schema pSchema){ - Schema unionSchema = pSchema.getTypes().get(pSchemaPos); - // default union element - if ( unionSchema == null ) { - unionSchema = pSchema.getTypes().get(CassandraStore.DEFAULT_UNION_SCHEMA); - } - return unionSchema; - } - - public void setValue(HColumn<ByteBuffer, ByteBuffer> hColumn) { - this.hColumn = hColumn; - } -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSuperColumn.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSuperColumn.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSuperColumn.java deleted file mode 100644 index 2b64705..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSuperColumn.java +++ /dev/null @@ -1,197 +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.gora.cassandra.query; - -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import me.prettyprint.cassandra.serializers.IntegerSerializer; -import me.prettyprint.cassandra.serializers.StringSerializer; -import me.prettyprint.hector.api.beans.HColumn; -import me.prettyprint.hector.api.beans.HSuperColumn; - -import org.apache.avro.Schema; -import org.apache.avro.Schema.Field; -import org.apache.avro.Schema.Type; -import org.apache.gora.cassandra.serializers.CharSequenceSerializer; -import org.apache.gora.cassandra.store.CassandraStore; -import org.apache.gora.persistency.impl.PersistentBase; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public class CassandraSuperColumn extends CassandraColumn { - public static final Logger LOG = LoggerFactory.getLogger(CassandraSuperColumn.class); - - private HSuperColumn<String, ByteBuffer, ByteBuffer> hSuperColumn; - - public ByteBuffer getName() { - return StringSerializer.get().toByteBuffer(hSuperColumn.getName()); - } - - private Object getSuperValue(Field field, Schema fieldSchema, Type type){ - Object value = null; - - switch (type) { - case ARRAY: - List<Object> array = new ArrayList<>(); - - for (HColumn<ByteBuffer, ByteBuffer> hColumn : this.hSuperColumn.getColumns()) { - Object memberValue = fromByteBuffer(fieldSchema.getElementType(), hColumn.getValue()); - // int i = IntegerSerializer().get().fromByteBuffer(hColumn.getName()); - array.add(memberValue); - } - value = array; - - break; - case MAP: - Map<CharSequence, Object> map = new HashMap<>(); - - for (HColumn<ByteBuffer, ByteBuffer> hColumn : this.hSuperColumn.getColumns()) { - CharSequence mapKey = CharSequenceSerializer.get().fromByteBuffer(hColumn.getName()); - if (!mapKey.toString().contains(CassandraStore.UNION_COL_SUFIX)) { - Object memberValue = null; - // We need detect real type for UNION Fields - if (fieldSchema.getValueType().getType().equals(Type.UNION)){ - - HColumn<ByteBuffer, ByteBuffer> cc = getUnionTypeColumn(mapKey - + CassandraStore.UNION_COL_SUFIX, this.hSuperColumn.getColumns()); - Integer unionIndex = getUnionIndex(cc); - Schema realSchema = fieldSchema.getValueType().getTypes().get(unionIndex); - memberValue = fromByteBuffer(realSchema, hColumn.getValue()); - - }else{ - memberValue = fromByteBuffer(fieldSchema.getValueType(), hColumn.getValue()); - } - map.put(mapKey, memberValue); - } - } - value = map; - - break; - case RECORD: - String fullName = fieldSchema.getFullName(); - - Class<?> claz = null; - try { - claz = Class.forName(fullName); - } catch (ClassNotFoundException cnfe) { - LOG.warn("Unable to load class " + fullName, cnfe); - break; - } - - try { - value = claz.newInstance(); - } catch (InstantiationException ie) { - LOG.warn("Instantiation error", ie); - break; - } catch (IllegalAccessException iae) { - LOG.warn("Illegal access error", iae); - break; - } - - // we updated the value instance, now update its members - if (value instanceof PersistentBase) { - PersistentBase record = (PersistentBase) value; - - for (HColumn<ByteBuffer, ByteBuffer> hColumn : this.hSuperColumn.getColumns()) { - String memberName = StringSerializer.get().fromByteBuffer(hColumn.getName()); - - if (memberName == null || memberName.length() == 0) { - LOG.warn("member name is null or empty."); - continue; - } - - if (!memberName.contains(CassandraStore.UNION_COL_SUFIX)) { - Field memberField = fieldSchema.getField(memberName); - Schema memberSchema = memberField.schema(); - Type memberType = memberSchema.getType(); - - CassandraSubColumn cassandraColumn = new CassandraSubColumn(); - cassandraColumn.setField(memberField); - cassandraColumn.setValue(hColumn); - - if (memberType.equals(Type.UNION)){ - HColumn<ByteBuffer, ByteBuffer> hc = getUnionTypeColumn(memberField.name() - + CassandraStore.UNION_COL_SUFIX, this.hSuperColumn.getColumns().toArray()); - Integer unionIndex = getUnionIndex(hc); - cassandraColumn.setUnionType(unionIndex); - } - - record.put(record.getSchema().getField(memberName).pos(), cassandraColumn.getValue()); - } - } - } - break; - case UNION: - int schemaPos = this.getUnionType(); - Schema unioSchema = fieldSchema.getTypes().get(schemaPos); - Type unionType = unioSchema.getType(); - value = getSuperValue(field, unioSchema, unionType); - break; - default: - Object memberValue = null; - // Using for UnionIndex of Union type field get value. UnionIndex always Integer. - for (HColumn<ByteBuffer, ByteBuffer> hColumn : this.hSuperColumn.getColumns()) { - memberValue = fromByteBuffer(fieldSchema, hColumn.getValue()); - } - value = memberValue; - LOG.warn("Type: " + type.name() + " not supported for field: " + field.name()); - } - return value; - } - - private Integer getUnionIndex(HColumn<ByteBuffer, ByteBuffer> uc){ - Integer val = IntegerSerializer.get().fromByteBuffer(uc.getValue()); - return Integer.parseInt(val.toString()); - } - - private HColumn<ByteBuffer, ByteBuffer> getUnionTypeColumn(String fieldName, - List<HColumn<ByteBuffer, ByteBuffer>> columns) { - return getUnionTypeColumn(fieldName, columns.toArray()); -} - - private HColumn<ByteBuffer, ByteBuffer> getUnionTypeColumn(String fieldName, Object[] hColumns) { - for (Object currentHColumn : hColumns) { - @SuppressWarnings("unchecked") - HColumn<ByteBuffer, ByteBuffer> hColumn = (HColumn<ByteBuffer, ByteBuffer>) currentHColumn; - String columnName = StringSerializer.get().fromByteBuffer(hColumn.getNameBytes().duplicate()); - if (fieldName.equals(columnName)) - return hColumn; - } - return null; -} - - public Object getValue() { - Field field = getField(); - Schema fieldSchema = field.schema(); - Type type = fieldSchema.getType(); - - Object value = getSuperValue(field, fieldSchema, type); - - return value; - } - - public void setValue(HSuperColumn<String, ByteBuffer, ByteBuffer> hSuperColumn) { - this.hSuperColumn = hSuperColumn; - } - -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/package-info.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/package-info.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/package-info.java deleted file mode 100644 index 49faefa..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/package-info.java +++ /dev/null @@ -1,21 +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. - */ -/** - * This package contains all the Cassandra store query representation class as well as Result set representing class - * when query is executed over the Cassandra dataStore. - */ -package org.apache.gora.cassandra.query; \ No newline at end of file http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializerUtil.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializerUtil.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializerUtil.java deleted file mode 100644 index 1fa6883..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializerUtil.java +++ /dev/null @@ -1,94 +0,0 @@ -package org.apache.gora.cassandra.serializers; - -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.util.concurrent.ConcurrentHashMap; - -import org.apache.avro.Schema; -import org.apache.avro.io.BinaryDecoder; -import org.apache.avro.io.BinaryEncoder; -import org.apache.avro.io.DecoderFactory; -import org.apache.avro.io.EncoderFactory; -import org.apache.avro.specific.SpecificDatumReader; -import org.apache.avro.specific.SpecificDatumWriter; - -public class AvroSerializerUtil { - - /** - * Threadlocals maintaining reusable binary decoders and encoders. - */ - private static ThreadLocal<ByteArrayOutputStream> outputStream = - new ThreadLocal<>(); - - public static final ThreadLocal<BinaryEncoder> encoders = - new ThreadLocal<>(); - - public static final ThreadLocal<BinaryDecoder> decoders = - new ThreadLocal<>(); - - /** - * Create a {@link java.util.concurrent.ConcurrentHashMap} for the - * datum readers and writers. - * This is necessary because they are not thread safe, at least not before - * Avro 1.4.0 (See AVRO-650). - * When they are thread safe, it is possible to maintain a single reader and - * writer pair for every schema, instead of one for every thread. - * @see <a href="https://issues.apache.org/jira/browse/AVRO-650";>AVRO-650</a> - */ - public static final ConcurrentHashMap<String, SpecificDatumWriter<?>> writerMap = - new ConcurrentHashMap<>(); - - public static final ConcurrentHashMap<String, SpecificDatumReader<?>> readerMap = - new ConcurrentHashMap<>(); - - @SuppressWarnings({ "unchecked", "rawtypes" }) - public static <T> byte[] serializer(T value, Schema schema) throws IOException{ - SpecificDatumWriter writer = writerMap.get(schema.getFullName()); - if (writer == null) { - writer = new SpecificDatumWriter(schema);// ignore dirty bits - writerMap.put(schema.getFullName(),writer); - } - - BinaryEncoder encoderFromCache = encoders.get(); - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - outputStream.set(bos); - BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(bos, null); - if (encoderFromCache == null) { - encoders.set(encoder); - } - - //reset the buffers - ByteArrayOutputStream os = outputStream.get(); - os.reset(); - - writer.write(value, encoder); - encoder.flush(); - byte[] byteValue = os.toByteArray(); - return byteValue; - } - - public static Object deserializer(Object value, Schema schema) throws IOException{ - String schemaId = schema.getFullName(); - - SpecificDatumReader<?> reader = readerMap.get(schemaId); - if (reader == null) { - reader = new SpecificDatumReader(schema);// ignore dirty bits - SpecificDatumReader localReader=null; - if((localReader=readerMap.putIfAbsent(schemaId, reader))!=null) { - reader = localReader; - } - } - - // initialize a decoder, possibly reusing previous one - BinaryDecoder decoderFromCache = decoders.get(); - BinaryDecoder decoder = DecoderFactory.get().binaryDecoder((byte[])value, null); - // put in threadlocal cache if the initial get was empty - if (decoderFromCache==null) { - decoders.set(decoder); - } - - Object result = reader.read(null, decoder); - return result; - - } -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CharSequenceSerializer.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CharSequenceSerializer.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CharSequenceSerializer.java deleted file mode 100644 index 5afb2e9..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CharSequenceSerializer.java +++ /dev/null @@ -1,64 +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.gora.cassandra.serializers; - -import static me.prettyprint.hector.api.ddl.ComparatorType.UTF8TYPE; - -import java.nio.ByteBuffer; - -import org.apache.avro.util.Utf8; - -import me.prettyprint.cassandra.serializers.AbstractSerializer; -import me.prettyprint.cassandra.serializers.StringSerializer; -import me.prettyprint.hector.api.ddl.ComparatorType; - -/** - * A CharSequenceSerializer translates the byte[] to and from CharSequenceSerializer object of Avro. - */ -public final class CharSequenceSerializer extends AbstractSerializer<CharSequence> { - - private static final CharSequenceSerializer instance = new CharSequenceSerializer(); - - public static CharSequenceSerializer get() { - return instance; - } - - @Override - public ByteBuffer toByteBuffer(CharSequence obj) { - if (obj == null) { - return null; - } - return StringSerializer.get().toByteBuffer(obj.toString()); - } - - @Override - //TODO: CharSequence cause Test Fail. All tests set UTF8. When change test set type. This will be CharSequence - public Utf8 fromByteBuffer(ByteBuffer byteBuffer) { - if (byteBuffer == null) { - return null; - } - return new Utf8(StringSerializer.get().fromByteBuffer(byteBuffer)); - } - - @Override - public ComparatorType getComparatorType() { - return UTF8TYPE; - } - -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/GoraSerializerTypeInferer.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/GoraSerializerTypeInferer.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/GoraSerializerTypeInferer.java deleted file mode 100644 index e9521c8..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/GoraSerializerTypeInferer.java +++ /dev/null @@ -1,230 +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.gora.cassandra.serializers; - -import java.nio.ByteBuffer; -import java.util.Map; - -import me.prettyprint.cassandra.serializers.BooleanSerializer; -import me.prettyprint.cassandra.serializers.ByteBufferSerializer; -import me.prettyprint.cassandra.serializers.BytesArraySerializer; -import me.prettyprint.cassandra.serializers.DoubleSerializer; -import me.prettyprint.cassandra.serializers.FloatSerializer; -import me.prettyprint.cassandra.serializers.IntegerSerializer; -import me.prettyprint.cassandra.serializers.LongSerializer; -import me.prettyprint.cassandra.serializers.ObjectSerializer; -import me.prettyprint.cassandra.serializers.SerializerTypeInferer; -import me.prettyprint.cassandra.serializers.StringSerializer; -import me.prettyprint.hector.api.Serializer; - -import org.apache.avro.Schema; -import org.apache.avro.Schema.Type; -import org.apache.avro.generic.GenericArray; -import org.apache.avro.specific.SpecificFixed; -import org.apache.avro.util.Utf8; -import org.apache.gora.persistency.Persistent; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Utility class that infers the concrete Serializer needed to turn a value into - * its binary representation - */ -public class GoraSerializerTypeInferer { - - public static final Logger LOG = LoggerFactory.getLogger(GoraSerializerTypeInferer.class); - - @SuppressWarnings({ "rawtypes", "unchecked" }) - public static <T> Serializer<T> getSerializer(Object value) { - Serializer serializer = null; - if (value == null) { - serializer = ByteBufferSerializer.get(); - } else if (value instanceof CharSequence) { - serializer = CharSequenceSerializer.get(); - } else if (value instanceof Utf8) { - serializer = CharSequenceSerializer.get(); - } else if (value instanceof Boolean) { - serializer = BooleanSerializer.get(); - } else if (value instanceof ByteBuffer) { - serializer = ByteBufferSerializer.get(); - } else if (value instanceof byte[]) { - serializer = BytesArraySerializer.get(); - } else if (value instanceof Double) { - serializer = DoubleSerializer.get(); - } else if (value instanceof Float) { - serializer = FloatSerializer.get(); - } else if (value instanceof Integer) { - serializer = IntegerSerializer.get(); - } else if (value instanceof Long) { - serializer = LongSerializer.get(); - } else if (value instanceof String) { - serializer = StringSerializer.get(); - } else if (value instanceof SpecificFixed) { - serializer = SpecificFixedSerializer.get(value.getClass()); - } else if (value instanceof GenericArray) { - Schema schema = ((GenericArray)value).getSchema(); - if (schema.getType() == Type.ARRAY) { - schema = schema.getElementType(); - } - serializer = ListSerializer.get(schema); - } else if (value instanceof Map) { - Map map = (Map)value; - if (map.size() == 0) { - serializer = ByteBufferSerializer.get(); - } - else { - Object value0 = map.values().iterator().next(); - Schema schema = TypeUtils.getSchema(value0); - serializer = MapSerializer.get(schema); - } - } else if (value instanceof Persistent){ - serializer = ObjectSerializer.get(); - } - else { - serializer = SerializerTypeInferer.getSerializer(value); - } - return serializer; - } - - @SuppressWarnings({ "rawtypes", "unchecked" }) - public static <T> Serializer<T> getSerializer(Class<?> valueClass) { - Serializer serializer = null; - if (valueClass.equals(Utf8.class) || valueClass.equals(CharSequence.class)) { - serializer = CharSequenceSerializer.get(); - } else if (valueClass.equals(Boolean.class) || valueClass.equals(boolean.class)) { - serializer = BooleanSerializer.get(); - } else if (valueClass.equals(ByteBuffer.class)) { - serializer = ByteBufferSerializer.get(); - } else if (valueClass.equals(Double.class) || valueClass.equals(double.class)) { - serializer = DoubleSerializer.get(); - } else if (valueClass.equals(Float.class) || valueClass.equals(float.class)) { - serializer = FloatSerializer.get(); - } else if (valueClass.equals(Integer.class) || valueClass.equals(int.class)) { - serializer = IntegerSerializer.get(); - } else if (valueClass.equals(Long.class) || valueClass.equals(long.class)) { - serializer = LongSerializer.get(); - } else if (valueClass.equals(String.class)) { - serializer = StringSerializer.get(); - } else { - serializer = SerializerTypeInferer.getSerializer(valueClass); - } - return serializer; - } - - @SuppressWarnings({ "rawtypes", "unchecked" }) - public static <T> Serializer<T> getSerializer(Schema schema) { - Serializer serializer = null; - Type type = schema.getType(); - if (type.equals(Type.STRING)) { - serializer = CharSequenceSerializer.get(); - } else if (type.equals(Type.BOOLEAN)) { - serializer = BooleanSerializer.get(); - } else if (type.equals(Type.BYTES)) { - serializer = ByteBufferSerializer.get(); - } else if (type.equals(Type.DOUBLE)) { - serializer = DoubleSerializer.get(); - } else if (type.equals(Type.FLOAT)) { - serializer = FloatSerializer.get(); - } else if (type.equals(Type.INT)) { - serializer = IntegerSerializer.get(); - } else if (type.equals(Type.LONG)) { - serializer = LongSerializer.get(); - } else if (type.equals(Type.FIXED)) { - Class clazz = TypeUtils.getClass(schema); - serializer = SpecificFixedSerializer.get(clazz); - // serializer = SpecificFixedSerializer.get(schema); - } else if (type.equals(Type.ARRAY)) { - serializer = ListSerializer.get(schema.getElementType()); - } else if (type.equals(Type.MAP)) { - serializer = MapSerializer.get(schema.getValueType()); - } else if (type.equals(Type.UNION)){ - serializer = ByteBufferSerializer.get(); - } else if (type.equals(Type.RECORD)){ - serializer = BytesArraySerializer.get(); - } else { - serializer = null; - } - return serializer; - } - - @SuppressWarnings({ "rawtypes", "unchecked" }) - public static <T> Serializer<T> getSerializer(Type type) { - Serializer serializer = null; - if (type == Type.STRING) { - serializer = CharSequenceSerializer.get(); - } else if (type == Type.BOOLEAN) { - serializer = BooleanSerializer.get(); - } else if (type == Type.BYTES) { - serializer = ByteBufferSerializer.get(); - } else if (type == Type.DOUBLE) { - serializer = DoubleSerializer.get(); - } else if (type == Type.FLOAT) { - serializer = FloatSerializer.get(); - } else if (type == Type.INT) { - serializer = IntegerSerializer.get(); - } else if (type == Type.LONG) { - serializer = LongSerializer.get(); - } else if (type == Type.FIXED) { - serializer = SpecificFixedSerializer.get(); - } else { - serializer = null; - } - return serializer; - } - - @SuppressWarnings({ "rawtypes", "unchecked" }) - public static <T> Serializer<T> getSerializer(Type type, Type elementType) { - Serializer serializer = null; - if (type == null) { - if (elementType == null) { - serializer = null; - } else { - serializer = getSerializer(elementType); - } - } else { - if (elementType == null) { - serializer = getSerializer(type); - } - } - - if (type == Type.ARRAY) { - serializer = ListSerializer.get(elementType); - } else if (type == Type.MAP) { - serializer = MapSerializer.get(elementType); - } else { - serializer = null; - } - return serializer; - } - - @SuppressWarnings({ "rawtypes", "unchecked" }) - public static <T> Serializer<T> getSerializer(Type type, Class<T> clazz) { - Serializer serializer = null; - if (type != Type.FIXED) { - serializer = null; - } - if (clazz == null) { - serializer = null; - } else { - serializer = SpecificFixedSerializer.get(clazz); - } - return serializer; - } - -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/ListSerializer.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/ListSerializer.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/ListSerializer.java deleted file mode 100644 index 6422082..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/ListSerializer.java +++ /dev/null @@ -1,193 +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.gora.cassandra.serializers; - -import java.nio.BufferUnderflowException; -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import me.prettyprint.cassandra.serializers.AbstractSerializer; -import me.prettyprint.cassandra.serializers.BytesArraySerializer; -import me.prettyprint.cassandra.serializers.IntegerSerializer; -import me.prettyprint.hector.api.Serializer; -import me.prettyprint.hector.api.ddl.ComparatorType; - -import org.apache.avro.Schema; -import org.apache.avro.Schema.Type; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * A GenericArraySerializer translates the byte[] to and from GenericArray of Avro. - */ -public class ListSerializer<T> extends AbstractSerializer<List<T>> { - - public static final Logger LOG = LoggerFactory.getLogger(ListSerializer.class); - - private static Map<Type, ListSerializer> elementTypeToSerializerMap = new HashMap<>(); - private static Map<Class, ListSerializer> fixedClassToSerializerMap = new HashMap<>(); - - public static ListSerializer get(Type elementType) { - ListSerializer serializer = elementTypeToSerializerMap.get(elementType); - if (serializer == null) { - serializer = new ListSerializer(elementType); - elementTypeToSerializerMap.put(elementType, serializer); - } - return serializer; - } - - public static ListSerializer get(Type elementType, Class clazz) { - if (elementType != Type.FIXED) { - return null; - } - ListSerializer serializer = elementTypeToSerializerMap.get(clazz); - if (serializer == null) { - serializer = new ListSerializer(clazz); - fixedClassToSerializerMap.put(clazz, serializer); - } - return serializer; - } - - public static ListSerializer get(Schema elementSchema) { - Type type = elementSchema.getType(); - if (type == Type.FIXED) { - return get(Type.FIXED, TypeUtils.getClass(elementSchema)); - } else { - return get(type); - } - } - - private Schema elementSchema = null; - private Type elementType = null; - private int size = -1; - private Class<T> clazz = null; - private Serializer<T> elementSerializer = null; - - public ListSerializer(Serializer<T> elementSerializer) { - this.elementSerializer = elementSerializer; - } - - public ListSerializer(Schema elementSchema) { - this.elementSchema = elementSchema; - elementType = elementSchema.getType(); - size = TypeUtils.getFixedSize(elementSchema); - elementSerializer = GoraSerializerTypeInferer.getSerializer(elementSchema); - } - - @SuppressWarnings("unchecked") - public ListSerializer(Type elementType) { - this.elementType = elementType; - if (elementType != Type.FIXED) { - elementSchema = Schema.create(elementType); - } - clazz = (Class<T>) TypeUtils.getClass(elementType); - size = TypeUtils.getFixedSize(elementType); - elementSerializer = GoraSerializerTypeInferer.getSerializer(elementType); - } - - public ListSerializer(Class<T> clazz) { - this.clazz = clazz; - elementType = TypeUtils.getType(clazz); - size = TypeUtils.getFixedSize(clazz); - if (elementType == null || elementType == Type.FIXED) { - elementType = Type.FIXED; - elementSchema = TypeUtils.getSchema(clazz); - elementSerializer = GoraSerializerTypeInferer.getSerializer(elementType, clazz); - } else { - elementSerializer = GoraSerializerTypeInferer.getSerializer(elementType); - } - } - - @Override - public ByteBuffer toByteBuffer(List<T> array) { - if (array == null) { - return null; - } - if (size > 0) { - return toByteBufferWithFixedLengthElements(array); - } else { - return toByteBufferWithVariableLengthElements(array); - } - } - - private ByteBuffer toByteBufferWithFixedLengthElements(List<T> array) { - ByteBuffer byteBuffer = ByteBuffer.allocate(array.size() * size); - for (T element : array) { - byteBuffer.put(elementSerializer.toByteBuffer(element)); - } - byteBuffer.rewind(); - return byteBuffer; - } - - private ByteBuffer toByteBufferWithVariableLengthElements(List<T> array) { - int n = array.size(); - List<byte[]> list = new ArrayList<>(n); - n *= 4; - for (T element : array) { - byte[] bytes = BytesArraySerializer.get().fromByteBuffer(elementSerializer.toByteBuffer(element)); - list.add(bytes); - n += bytes.length; - } - ByteBuffer byteBuffer = ByteBuffer.allocate(n); - for (byte[] bytes : list) { - byteBuffer.put(IntegerSerializer.get().toByteBuffer(bytes.length)); - byteBuffer.put(BytesArraySerializer.get().toByteBuffer(bytes)); - } - byteBuffer.rewind(); - return byteBuffer; - } - - @Override - public List<T> fromByteBuffer(ByteBuffer byteBuffer) { - if (byteBuffer == null) { - return null; - } - ArrayList<T> array = new ArrayList<>(); - while (true) { - T element = null; - try { - if (size > 0) { - element = elementSerializer.fromByteBuffer(byteBuffer); - } - else { - int n = IntegerSerializer.get().fromByteBuffer(byteBuffer); - byte[] bytes = new byte[n]; - byteBuffer.get(bytes, 0, n); - element = elementSerializer.fromByteBuffer( BytesArraySerializer.get().toByteBuffer(bytes) ); - } - } catch (BufferUnderflowException e) { - break; - } - if (element == null) { - break; - } - array.add(element); - } - return array; - } - - @Override - public ComparatorType getComparatorType() { - return elementSerializer.getComparatorType(); - } - -} http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/MapSerializer.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/MapSerializer.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/MapSerializer.java deleted file mode 100644 index 2bdcea0..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/MapSerializer.java +++ /dev/null @@ -1,223 +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.gora.cassandra.serializers; - -import java.nio.BufferUnderflowException; -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import me.prettyprint.cassandra.serializers.AbstractSerializer; -import me.prettyprint.cassandra.serializers.BytesArraySerializer; -import me.prettyprint.cassandra.serializers.IntegerSerializer; -import me.prettyprint.hector.api.Serializer; -import me.prettyprint.hector.api.ddl.ComparatorType; - -import org.apache.avro.Schema; -import org.apache.avro.Schema.Type; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * A MapSerializer translates the byte[] to and from Map of Avro. - */ -public class MapSerializer<T> extends AbstractSerializer<Map<CharSequence, T>> { - - public static final Logger LOG = LoggerFactory.getLogger(MapSerializer.class); - - private static Map<Type, MapSerializer> valueTypeToSerializerMap = new HashMap<>(); - private static Map<Class, MapSerializer> fixedClassToSerializerMap = new HashMap<>(); - - public static MapSerializer get(Type valueType) { - MapSerializer serializer = valueTypeToSerializerMap.get(valueType); - if (serializer == null) { - serializer = new MapSerializer(valueType); - valueTypeToSerializerMap.put(valueType, serializer); - } - return serializer; - } - - public static MapSerializer get(Type valueType, Class clazz) { - if (valueType != Type.FIXED) { - return null; - } - MapSerializer serializer = valueTypeToSerializerMap.get(clazz); - if (serializer == null) { - serializer = new MapSerializer(clazz); - fixedClassToSerializerMap.put(clazz, serializer); - } - return serializer; - } - - public static MapSerializer get(Schema valueSchema) { - Type type = valueSchema.getType(); - if (type == Type.FIXED) { - return get(Type.FIXED, TypeUtils.getClass(valueSchema)); - } else { - return get(type); - } - } - - private Schema valueSchema = null; - private Type valueType = null; - private int size = -1; - private Class<T> clazz = null; - private Serializer<T> valueSerializer = null; - - public MapSerializer(Serializer<T> valueSerializer) { - this.valueSerializer = valueSerializer; - } - - public MapSerializer(Schema valueSchema) { - this.valueSchema = valueSchema; - valueType = valueSchema.getType(); - size = TypeUtils.getFixedSize(valueSchema); - valueSerializer = GoraSerializerTypeInferer.getSerializer(valueSchema); - } - - @SuppressWarnings("unchecked") - public MapSerializer(Type valueType) { - this.valueType = valueType; - if (valueType != Type.FIXED) { - valueSchema = Schema.create(valueType); - } - clazz = (Class<T>) TypeUtils.getClass(valueType); - size = TypeUtils.getFixedSize(valueType); - valueSerializer = GoraSerializerTypeInferer.getSerializer(valueType); - } - - public MapSerializer(Class<T> clazz) { - this.clazz = clazz; - valueType = TypeUtils.getType(clazz); - size = TypeUtils.getFixedSize(clazz); - if (valueType == null || valueType == Type.FIXED) { - valueType = Type.FIXED; - valueSchema = TypeUtils.getSchema(clazz); - valueSerializer = GoraSerializerTypeInferer.getSerializer(valueType, clazz); - } else { - valueSerializer = GoraSerializerTypeInferer.getSerializer(valueType); - } - } - - @Override - public ByteBuffer toByteBuffer(Map<CharSequence, T> map) { - if (map == null) { - return null; - } - if (size > 0) { - return toByteBufferWithFixedLengthElements(map); - } else { - return toByteBufferWithVariableLengthElements(map); - } - } - - private ByteBuffer toByteBufferWithFixedLengthElements(Map<CharSequence, T> map) { - int n = map.size(); - List<byte[]> list = new ArrayList<>(n); - n *= 4; - for (CharSequence key : map.keySet()) { - T value = map.get(key); - byte[] bytes = BytesArraySerializer.get().fromByteBuffer(CharSequenceSerializer.get().toByteBuffer(key)); - list.add(bytes); - n += bytes.length; - bytes = BytesArraySerializer.get().fromByteBuffer(valueSerializer.toByteBuffer(value)); - list.add(bytes); - n += bytes.length; - } - ByteBuffer byteBuffer = ByteBuffer.allocate(n); - int i = 0; - for (byte[] bytes : list) { - if (i % 2 == 0) { - byteBuffer.put(IntegerSerializer.get().toByteBuffer(bytes.length)); - } - byteBuffer.put(BytesArraySerializer.get().toByteBuffer(bytes)); - i += 1; - } - byteBuffer.rewind(); - return byteBuffer; - } - - private ByteBuffer toByteBufferWithVariableLengthElements(Map<CharSequence, T> map) { - int n = map.size(); - List<byte[]> list = new ArrayList<>(n); - n *= 8; - for (CharSequence key : map.keySet()) { - T value = map.get(key); - byte[] bytes = BytesArraySerializer.get().fromByteBuffer(CharSequenceSerializer.get().toByteBuffer(key)); - list.add(bytes); - n += bytes.length; - bytes = BytesArraySerializer.get().fromByteBuffer(valueSerializer.toByteBuffer(value)); - list.add(bytes); - n += bytes.length; - } - ByteBuffer byteBuffer = ByteBuffer.allocate(n); - for (byte[] bytes : list) { - byteBuffer.put(IntegerSerializer.get().toByteBuffer(bytes.length)); - byteBuffer.put(BytesArraySerializer.get().toByteBuffer(bytes)); - } - byteBuffer.rewind(); - return byteBuffer; - } - - @Override - public Map<CharSequence, T> fromByteBuffer(ByteBuffer byteBuffer) { - if (byteBuffer == null) { - return null; - } - Map<CharSequence, T> map = new HashMap<>(); - while (true) { - CharSequence key = null; - T value = null; - try { - int n = IntegerSerializer.get().fromByteBuffer(byteBuffer); - byte[] bytes = new byte[n]; - byteBuffer.get(bytes, 0, n); - key = CharSequenceSerializer.get().fromByteBuffer( BytesArraySerializer.get().toByteBuffer(bytes) ); - - if (size > 0) { - value = valueSerializer.fromByteBuffer(byteBuffer); - } - else { - n = IntegerSerializer.get().fromByteBuffer(byteBuffer); - bytes = new byte[n]; - byteBuffer.get(bytes, 0, n); - value = valueSerializer.fromByteBuffer( BytesArraySerializer.get().toByteBuffer(bytes) ); - } - } catch (BufferUnderflowException e) { - break; - } - if (key == null) { - break; - } - if (value == null) { - break; - } - map.put(key, value); - } - return map; - } - - @Override - public ComparatorType getComparatorType() { - return valueSerializer.getComparatorType(); - } - -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/SpecificFixedSerializer.java ---------------------------------------------------------------------- diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/SpecificFixedSerializer.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/SpecificFixedSerializer.java deleted file mode 100644 index 71f935a..0000000 --- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/SpecificFixedSerializer.java +++ /dev/null @@ -1,119 +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.gora.cassandra.serializers; - -import java.nio.BufferUnderflowException; -import java.nio.ByteBuffer; -import java.util.HashMap; -import java.util.Map; - -import me.prettyprint.cassandra.serializers.AbstractSerializer; -import me.prettyprint.cassandra.serializers.BytesArraySerializer; -import me.prettyprint.hector.api.ddl.ComparatorType; -import static me.prettyprint.hector.api.ddl.ComparatorType.BYTESTYPE; - -import org.apache.avro.specific.SpecificFixed; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * A SpecificFixedSerializer translates the byte[] to and from SpecificFixed of Avro. - */ -public class SpecificFixedSerializer extends AbstractSerializer<SpecificFixed> { - - public static final Logger LOG = LoggerFactory.getLogger(SpecificFixedSerializer.class); - - // for toByteBuffer - private static SpecificFixedSerializer serializer = new SpecificFixedSerializer(SpecificFixed.class); - - // for fromByteBuffer, requiring Class info - public static SpecificFixedSerializer get() { - return serializer; - } - - private static Map<Class, SpecificFixedSerializer> classToSerializerMap = new HashMap<>(); - - public static SpecificFixedSerializer get(Class clazz) { - SpecificFixedSerializer serializer = classToSerializerMap.get(clazz); - if (serializer == null) { - serializer = new SpecificFixedSerializer(clazz); - classToSerializerMap.put(clazz, serializer); - } - return serializer; - } - - private Class<? extends SpecificFixed> clazz; - - public SpecificFixedSerializer(Class<? extends SpecificFixed> clazz) { - this.clazz = clazz; - } - - @Override - public ByteBuffer toByteBuffer(SpecificFixed fixed) { - if (fixed == null) { - return null; - } - byte[] bytes = fixed.bytes(); - if (bytes.length < 1) { - return null; - } - return BytesArraySerializer.get().toByteBuffer(bytes); - } - - @Override - public SpecificFixed fromByteBuffer(ByteBuffer byteBuffer) { - if (byteBuffer == null) { - return null; - } - - Object value = null; - try { - value = clazz.newInstance(); - } catch (InstantiationException ie) { - LOG.warn("Instantiation error for class=" + clazz, ie); - return null; - } catch (IllegalAccessException iae) { - LOG.warn("Illegal access error for class=" + clazz, iae); - return null; - } - - if (! (value instanceof SpecificFixed)) { - LOG.warn("Not an instance of SpecificFixed"); - return null; - } - - SpecificFixed fixed = (SpecificFixed) value; - byte[] bytes = fixed.bytes(); - try { - byteBuffer.get(bytes, 0, bytes.length); - } - catch (BufferUnderflowException e) { - // LOG.info(e.toString() + " : class=" + clazz.getName() + " length=" + bytes.length); - throw e; - } - fixed.bytes(bytes); - return fixed; - } - - @Override - public ComparatorType getComparatorType() { - return BYTESTYPE; - } - -}
