You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gora.apache.org by le...@apache.org on 2017/08/23 20:55:26 UTC
[28/37] gora git commit: Add tests for avro
Add tests for avro
Project: http://git-wip-us.apache.org/repos/asf/gora/repo
Commit: http://git-wip-us.apache.org/repos/asf/gora/commit/a9a3ad49
Tree: http://git-wip-us.apache.org/repos/asf/gora/tree/a9a3ad49
Diff: http://git-wip-us.apache.org/repos/asf/gora/diff/a9a3ad49
Branch: refs/heads/master
Commit: a9a3ad4922a45a76c8c6b1b76c011d72736e0c30
Parents: 962d7a6
Author: madhawa-gunasekara <ma...@wso2.com>
Authored: Fri Aug 4 21:35:14 2017 +0530
Committer: madhawa-gunasekara <ma...@wso2.com>
Committed: Thu Aug 10 23:45:56 2017 +0530
----------------------------------------------------------------------
gora-cassandra-cql/pom.xml | 5 +-
.../nativeSerialization/ComplexTypes.java | 85 ++-
.../generated/nativeSerialization/Customer.java | 81 +++
.../generated/nativeSerialization/Document.java | 137 +++++
.../generated/nativeSerialization/User.java | 86 ++-
.../org/apache/gora/cassandra/bean/Field.java | 1 +
.../persistent/CassandraNativePersistent.java | 108 ----
.../cassandra/query/CassandraResultSet.java | 6 +-
.../serializers/AvroCassandraUtils.java | 20 +-
.../cassandra/serializers/AvroSerializer.java | 126 +++--
.../serializers/CassandraQueryFactory.java | 526 +++++++++----------
.../serializers/CassandraSerializer.java | 87 +--
.../cassandra/serializers/NativeSerializer.java | 10 +-
.../gora/cassandra/store/CassandraMapping.java | 8 +-
.../store/CassandraMappingBuilder.java | 3 +
.../gora/cassandra/store/CassandraStore.java | 26 +-
.../conf/avroUDT/gora-cassandra-mapping.xml | 50 ++
.../src/test/conf/gora.properties | 4 +-
.../conf/nativeUDT/gora-cassandra-mapping.xml | 33 ++
.../gora/cassandra/GoraCassandraTestDriver.java | 8 +-
.../store/TestAvroSerializationWithUDT.java | 91 ++++
.../cassandra/store/TestCassandraStore.java | 97 +++-
.../TestCassandraStoreWithCassandraKey.java | 68 +++
.../store/TestNativeSerializationWithUDT.java | 91 ++++
pom.xml | 4 +-
25 files changed, 1248 insertions(+), 513 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/pom.xml
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/pom.xml b/gora-cassandra-cql/pom.xml
index ed98a18..b78db24 100644
--- a/gora-cassandra-cql/pom.xml
+++ b/gora-cassandra-cql/pom.xml
@@ -167,11 +167,10 @@
</exclusions>
</dependency>
- <dependency>
+<!-- <dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
- <version>4.0.37.Final</version>
- </dependency>
+ </dependency>-->
<dependency>
<groupId>org.apache.gora</groupId>
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/ComplexTypes.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/ComplexTypes.java b/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/ComplexTypes.java
index 797ea62..c161ef9 100644
--- a/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/ComplexTypes.java
+++ b/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/ComplexTypes.java
@@ -3,7 +3,10 @@ package org.apache.gora.cassandra.example.generated.nativeSerialization;
import com.datastax.driver.mapping.annotations.Column;
import com.datastax.driver.mapping.annotations.PartitionKey;
import com.datastax.driver.mapping.annotations.Table;
-import org.apache.gora.cassandra.persistent.CassandraNativePersistent;
+import com.datastax.driver.mapping.annotations.Transient;
+import org.apache.avro.Schema;
+import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.Tombstone;
import java.util.List;
import java.util.Map;
@@ -18,7 +21,7 @@ import java.util.UUID;
writeConsistency = "QUORUM",
caseSensitiveKeyspace = false,
caseSensitiveTable = true)
-public class ComplexTypes extends CassandraNativePersistent {
+public class ComplexTypes implements Persistent {
@Column
private List<String> listDataType;
@@ -98,4 +101,82 @@ public class ComplexTypes extends CassandraNativePersistent {
public void setId(String id) {
this.id = id;
}
+
+ @Transient
+ @Override
+ public void clear() {
+
+ }
+
+ @Transient
+ @Override
+ public boolean isDirty(int fieldIndex) {
+ return false;
+ }
+
+ @Transient
+ @Override
+ public boolean isDirty(String field) {
+ return false;
+ }
+
+ @Transient
+ @Override
+ public void setDirty() {
+
+ }
+
+ @Transient
+ @Override
+ public void setDirty(int fieldIndex) {
+
+ }
+
+ @Transient
+ @Override
+ public void clearDirty(int fieldIndex) {
+
+ }
+
+ @Transient
+ @Override
+ public void clearDirty(String field) {
+
+ }
+
+ @Transient
+ @Override
+ public Tombstone getTombstone() {
+ return null;
+ }
+
+ @Transient
+ @Override
+ public List<Schema.Field> getUnmanagedFields() {
+ return null;
+ }
+
+ @Transient
+ @Override
+ public Persistent newInstance() {
+ return new ComplexTypes();
+ }
+
+ @Transient
+ @Override
+ public boolean isDirty() {
+ return false;
+ }
+
+ @Transient
+ @Override
+ public void setDirty(String field) {
+
+ }
+
+ @Transient
+ @Override
+ public void clearDirty() {
+
+ }
}
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/Customer.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/Customer.java b/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/Customer.java
new file mode 100644
index 0000000..1b2de10
--- /dev/null
+++ b/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/Customer.java
@@ -0,0 +1,81 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.example.generated.nativeSerialization;
+
+import com.datastax.driver.mapping.annotations.Field;
+import com.datastax.driver.mapping.annotations.UDT;
+
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+@UDT(name = "customer", keyspace = "nativeTestKeySpace")
+public class Customer {
+
+ public Customer() {
+
+ }
+ @Field(name = "id")
+ private String id;
+
+ @Field(name = "name")
+ private String name;
+
+ @Field
+ private UUID age;
+
+ @Field(name = "coupon_code")
+ private String couponCode;
+
+ @Field(name = "address")
+ private String address;
+
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getCouponCode() {
+ return couponCode;
+ }
+
+ public void setCouponCode(String couponCode) {
+ this.couponCode = couponCode;
+ }
+
+ public String getAddress() {
+ return address;
+ }
+
+ public void setAddress(String address) {
+ this.address = address;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/Document.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/Document.java b/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/Document.java
new file mode 100644
index 0000000..f841a4e
--- /dev/null
+++ b/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/Document.java
@@ -0,0 +1,137 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.example.generated.nativeSerialization;
+
+import com.datastax.driver.mapping.annotations.Column;
+import com.datastax.driver.mapping.annotations.PartitionKey;
+import com.datastax.driver.mapping.annotations.Table;
+import com.datastax.driver.mapping.annotations.Transient;
+import org.apache.avro.Schema;
+import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.Tombstone;
+
+import java.util.List;
+
+@Table(keyspace = "nativeTestKeySpace", name = "documents",
+ readConsistency = "QUORUM",
+ writeConsistency = "QUORUM",
+ caseSensitiveKeyspace = false,
+ caseSensitiveTable = false)
+public class Document implements Persistent {
+ @Column
+ Customer customer;
+ @PartitionKey
+ @Column
+ String defaultId;
+
+ public String getDefaultId() {
+ return defaultId;
+ }
+
+ public void setDefaultId(String defaultId) {
+ this.defaultId = defaultId;
+ }
+
+ public Customer getCustomer() {
+ return customer;
+ }
+
+ public void setCustomer(Customer customer) {
+ this.customer = customer;
+ }
+
+ @Transient
+ @Override
+ public void clear() {
+
+ }
+
+ @Transient
+ @Override
+ public boolean isDirty(int fieldIndex) {
+ return false;
+ }
+
+ @Transient
+ @Override
+ public boolean isDirty(String field) {
+ return false;
+ }
+
+ @Transient
+ @Override
+ public void setDirty() {
+
+ }
+
+ @Transient
+ @Override
+ public void setDirty(int fieldIndex) {
+
+ }
+
+ @Transient
+ @Override
+ public void clearDirty(int fieldIndex) {
+
+ }
+
+ @Transient
+ @Override
+ public void clearDirty(String field) {
+
+ }
+
+ @Transient
+ @Override
+ public Tombstone getTombstone() {
+ return null;
+ }
+
+ @Transient
+ @Override
+ public List<Schema.Field> getUnmanagedFields() {
+ return null;
+ }
+
+ @Transient
+ @Override
+ public Persistent newInstance() {
+ return new Document();
+ }
+
+ @Transient
+ @Override
+ public boolean isDirty() {
+ return false;
+ }
+
+ @Transient
+ @Override
+ public void setDirty(String field) {
+
+ }
+
+ @Transient
+ @Override
+ public void clearDirty() {
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/User.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/User.java b/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/User.java
index c8d7a78..08abcd1 100644
--- a/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/User.java
+++ b/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/User.java
@@ -20,9 +20,13 @@ package org.apache.gora.cassandra.example.generated.nativeSerialization;
import com.datastax.driver.mapping.annotations.Column;
import com.datastax.driver.mapping.annotations.PartitionKey;
import com.datastax.driver.mapping.annotations.Table;
-import org.apache.gora.cassandra.persistent.CassandraNativePersistent;
+import com.datastax.driver.mapping.annotations.Transient;
+import org.apache.avro.Schema;
+import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.Tombstone;
import java.util.Date;
+import java.util.List;
import java.util.UUID;
/**
@@ -33,7 +37,7 @@ import java.util.UUID;
writeConsistency = "QUORUM",
caseSensitiveKeyspace = false,
caseSensitiveTable = false)
-public class User extends CassandraNativePersistent {
+public class User implements Persistent {
@PartitionKey
@Column(name = "user_id")
private UUID userId;
@@ -75,4 +79,82 @@ public class User extends CassandraNativePersistent {
public void setDateOfBirth(Date dateOfBirth) {
this.dateOfBirth = dateOfBirth;
}
+
+ @Transient
+ @Override
+ public void clear() {
+
+ }
+
+ @Transient
+ @Override
+ public boolean isDirty(int fieldIndex) {
+ return false;
+ }
+
+ @Transient
+ @Override
+ public boolean isDirty(String field) {
+ return false;
+ }
+
+ @Transient
+ @Override
+ public void setDirty() {
+
+ }
+
+ @Transient
+ @Override
+ public void setDirty(int fieldIndex) {
+
+ }
+
+ @Transient
+ @Override
+ public void clearDirty(int fieldIndex) {
+
+ }
+
+ @Transient
+ @Override
+ public void clearDirty(String field) {
+
+ }
+
+ @Transient
+ @Override
+ public Tombstone getTombstone() {
+ return null;
+ }
+
+ @Transient
+ @Override
+ public List<Schema.Field> getUnmanagedFields() {
+ return null;
+ }
+
+ @Transient
+ @Override
+ public Persistent newInstance() {
+ return new User();
+ }
+
+ @Transient
+ @Override
+ public boolean isDirty() {
+ return false;
+ }
+
+ @Transient
+ @Override
+ public void setDirty(String field) {
+
+ }
+
+ @Transient
+ @Override
+ public void clearDirty() {
+
+ }
}
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/Field.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/Field.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/Field.java
index 3bbda6d..237601d 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/Field.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/Field.java
@@ -30,6 +30,7 @@ public class Field {
private String columnName;
private String type;
+
private Map<String, String> properties;
public Field() {
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/persistent/CassandraNativePersistent.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/persistent/CassandraNativePersistent.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/persistent/CassandraNativePersistent.java
deleted file mode 100644
index 9d6e103..0000000
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/persistent/CassandraNativePersistent.java
+++ /dev/null
@@ -1,108 +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.persistent;
-
-import com.datastax.driver.mapping.annotations.Transient;
-import org.apache.avro.Schema;
-import org.apache.gora.persistency.Persistent;
-import org.apache.gora.persistency.Tombstone;
-
-import java.util.List;
-
-/**
- * This class should be used with Native Cassandra Serialization.
- */
-public abstract class CassandraNativePersistent implements Persistent {
- @Transient
- @Override
- public void clear() {
-
- }
-
- @Transient
- @Override
- public boolean isDirty(int fieldIndex) {
- return false;
- }
-
- @Transient
- @Override
- public boolean isDirty(String field) {
- return false;
- }
-
- @Transient
- @Override
- public void setDirty() {
-
- }
-
- @Transient
- @Override
- public void setDirty(int fieldIndex) {
-
- }
-
- @Transient
- @Override
- public void clearDirty(int fieldIndex) {
-
- }
-
- @Transient
- @Override
- public void clearDirty(String field) {
-
- }
-
- @Transient
- @Override
- public Tombstone getTombstone() {
- return null;
- }
-
- @Transient
- @Override
- public List<Schema.Field> getUnmanagedFields() {
- return null;
- }
-
- @Transient
- @Override
- public Persistent newInstance() {
- return null;
- }
-
- @Transient
- @Override
- public boolean isDirty() {
- return false;
- }
-
- @Transient
- @Override
- public void setDirty(String field) {
-
- }
-
- @Transient
- @Override
- public void clearDirty() {
-
- }
-}
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/CassandraResultSet.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/CassandraResultSet.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/CassandraResultSet.java
index c3b2e59..7ad106d 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/CassandraResultSet.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/CassandraResultSet.java
@@ -99,8 +99,10 @@ public class CassandraResultSet<K, T extends Persistent> extends ResultBase<K, T
}
/**
- * @param key
- * @param token
+ * This method adds Result Element into result lists, So when user retrieves values from the Result these objects will be passed.
+ *
+ * @param key key
+ * @param token persistent Object
*/
public void addResultElement(K key, T token) {
this.persistentKey.add(key);
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroCassandraUtils.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroCassandraUtils.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroCassandraUtils.java
index 7baa1b1..252cf7b 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroCassandraUtils.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroCassandraUtils.java
@@ -19,7 +19,6 @@
package org.apache.gora.cassandra.serializers;
import org.apache.avro.Schema;
-import org.apache.avro.specific.SpecificData;
import org.apache.avro.util.Utf8;
import org.apache.gora.cassandra.bean.CassandraKey;
import org.apache.gora.cassandra.bean.Field;
@@ -91,20 +90,7 @@ class AvroCassandraUtils {
switch (type) {
// Record can be persist with two ways, udt and bytes
case RECORD:
- PersistentBase persistent = (PersistentBase) fieldValue;
- if (field.getType().contains("frozen")) {
- PersistentBase newRecord = (PersistentBase) SpecificData.get().newRecord(persistent, persistent.getSchema());
- for (Schema.Field member : fieldSchema.getFields()) {
- if (member.pos() == 0 || !persistent.isDirty()) {
- continue;
- }
- Schema memberSchema = member.schema();
- Schema.Type memberType = memberSchema.getType();
- Object memberValue = persistent.get(member.pos());
- newRecord.put(member.pos(), getFieldValueFromAvroBean(memberSchema, memberType, memberValue, field));
- }
- fieldValue = newRecord;
- } else if (field.getType().contains("blob")) {
+ if (field.getType().contains("blob")) {
try {
byte[] serializedBytes = HBaseByteInterface.toBytes(fieldValue, fieldSchema);
fieldValue = ByteBuffer.wrap(serializedBytes);
@@ -112,13 +98,13 @@ class AvroCassandraUtils {
LOG.error("Error occurred when serializing {} field. {}", new Object[]{field.getFieldName(), e.getMessage()});
}
} else {
- throw new RuntimeException("");
+ throw new RuntimeException("Unsupported Data Type for Record, Currently Supported Data Types are blob and UDT for Records");
}
break;
case MAP:
Schema valueSchema = fieldSchema.getValueType();
Schema.Type valuetype = valueSchema.getType();
- HashMap<String, Object> map = new HashMap<>();
+ Map<String, Object> map = new HashMap<>();
for (Map.Entry<CharSequence, ?> e : ((Map<CharSequence, ?>) fieldValue).entrySet()) {
String mapKey = e.getKey().toString();
Object mapValue = e.getValue();
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializer.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializer.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializer.java
index 57d03f1..58b57dc 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializer.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializer.java
@@ -17,11 +17,15 @@
package org.apache.gora.cassandra.serializers;
+import com.datastax.driver.core.AbstractGettableData;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
+import com.datastax.driver.core.UDTValue;
+import com.datastax.driver.core.UserType;
import org.apache.avro.Schema;
+import org.apache.avro.specific.SpecificData;
import org.apache.commons.lang.ArrayUtils;
import org.apache.gora.cassandra.bean.CassandraKey;
import org.apache.gora.cassandra.bean.Field;
@@ -36,10 +40,16 @@ import org.apache.gora.store.DataStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Date;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
+import java.util.UUID;
/**
* This class contains the operations relates to Avro Serialization.
@@ -51,8 +61,8 @@ class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer {
private DataStore<K, T> cassandraDataStore;
- AvroSerializer(CassandraClient cassandraClient, DataStore<K, T> dataStore, CassandraMapping mapping) {
- super(cassandraClient, dataStore.getKeyClass(), dataStore.getPersistentClass(), mapping);
+ AvroSerializer(CassandraClient cassandraClient, DataStore<K, T> dataStore, CassandraMapping mapping, Schema schema) {
+ super(cassandraClient, dataStore.getKeyClass(), dataStore.getPersistentClass(), mapping, schema);
this.cassandraDataStore = dataStore;
}
@@ -71,7 +81,7 @@ class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer {
T obj = null;
if (iterator.hasNext()) {
obj = cassandraDataStore.newPersistent();
- Row row = iterator.next();
+ AbstractGettableData row = (AbstractGettableData) iterator.next();
populateValuesToPersistent(row, definitions, obj, fields);
}
return obj;
@@ -94,7 +104,35 @@ class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer {
}
if (persistent.isDirty(f.pos()) || mapping.getInlinedDefinedPartitionKey().equals(mapping.getFieldFromFieldName(fieldName))) {
Object value = persistentBase.get(f.pos());
- value = AvroCassandraUtils.getFieldValueFromAvroBean(f.schema(), f.schema().getType(), value, field);
+ String fieldType = field.getType();
+ if (fieldType.contains("frozen")) {
+ fieldType = fieldType.substring(fieldType.indexOf("<") + 1, fieldType.indexOf(">"));
+ UserType userType = client.getSession().getCluster().getMetadata().getKeyspace(mapping.getKeySpace().getName()).getUserType(fieldType);
+ UDTValue udtValue = userType.newValue();
+ Schema udtSchema = f.schema();
+ if (udtSchema.getType().equals(Schema.Type.UNION)) {
+ for (Schema schema : udtSchema.getTypes()) {
+ if (schema.getType().equals(Schema.Type.RECORD)) {
+ udtSchema = schema;
+ break;
+ }
+ }
+ }
+ PersistentBase udtObjectBase = (PersistentBase) value;
+ for (Schema.Field udtField : udtSchema.getFields()) {
+ Object udtFieldValue = AvroCassandraUtils.getFieldValueFromAvroBean(udtField.schema(), udtField.schema().getType(), udtObjectBase.get(udtField.name()), field);
+ if (udtField.schema().getType().equals(Schema.Type.MAP)) {
+ udtValue.setMap(udtField.name(), (Map) udtFieldValue);
+ } else if (udtField.schema().getType().equals(Schema.Type.ARRAY)) {
+ udtValue.setList(udtField.name(), (List) udtFieldValue);
+ } else {
+ udtValue.set(udtField.name(), udtFieldValue, (Class) udtFieldValue.getClass());
+ }
+ }
+ value = udtValue;
+ } else {
+ value = AvroCassandraUtils.getFieldValueFromAvroBean(f.schema(), f.schema().getType(), value, field);
+ }
values.add(value);
fields.add(fieldName);
}
@@ -122,7 +160,7 @@ class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer {
T obj = null;
if (iterator.hasNext()) {
obj = cassandraDataStore.newPersistent();
- Row row = iterator.next();
+ AbstractGettableData row = (AbstractGettableData) iterator.next();
populateValuesToPersistent(row, definitions, obj, mapping.getFieldNames());
}
return obj;
@@ -131,7 +169,7 @@ class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer {
/**
* This method wraps result set data in to DataEntry and creates a list of DataEntry.
**/
- private void populateValuesToPersistent(Row row, ColumnDefinitions columnDefinitions, PersistentBase base, String[] fields) {
+ private void populateValuesToPersistent(AbstractGettableData row, ColumnDefinitions columnDefinitions, PersistentBase base, String[] fields) {
Object paramValue;
for (String fieldName : fields) {
Schema.Field avroField = base.getSchema().getField(fieldName);
@@ -142,16 +180,15 @@ class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer {
}
Schema fieldSchema = avroField.schema();
String columnName = field.getColumnName();
- paramValue = getValue(row, columnDefinitions, columnName);
+ paramValue = getValue(row, columnDefinitions.getType(columnName), columnName, fieldSchema);
Object value = AvroCassandraUtils.getAvroFieldValue(paramValue, fieldSchema);
base.put(avroField.pos(), value);
}
}
- private Object getValue(Row row, ColumnDefinitions columnDefinitions, String columnName) {
+ private Object getValue(AbstractGettableData row, DataType columnType, String columnName, Schema schema) {
Object paramValue;
- Field field = mapping.getFieldFromColumnName(columnName);
- DataType columnType = columnDefinitions.getType(columnName);
+ String dataType;
switch (columnType.getName()) {
case ASCII:
paramValue = row.getString(columnName);
@@ -202,24 +239,33 @@ class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer {
paramValue = row.isNull(columnName) ? null : row.getUUID(columnName);
break;
case LIST:
- String dataType = field.getType();
- dataType = dataType.substring(dataType.indexOf("<") + 1, dataType.indexOf(">"));
+ dataType = columnType.getTypeArguments().get(0).toString();
paramValue = row.isNull(columnName) ? null : row.getList(columnName, getRelevantClassForCassandraDataType(dataType));
break;
case SET:
- dataType = field.getType();
- dataType = dataType.substring(dataType.indexOf("<") + 1, dataType.indexOf(">"));
+ dataType = columnType.getTypeArguments().get(0).toString();
paramValue = row.isNull(columnName) ? null : row.getList(columnName, getRelevantClassForCassandraDataType(dataType));
break;
case MAP:
- dataType = field.getType();
- dataType = dataType.substring(dataType.indexOf("<") + 1, dataType.indexOf(">"));
- dataType = dataType.split(",")[1];
+ dataType = columnType.getTypeArguments().get(1).toString();
// Avro supports only String for keys
paramValue = row.isNull(columnName) ? null : row.getMap(columnName, String.class, getRelevantClassForCassandraDataType(dataType));
break;
case UDT:
paramValue = row.isNull(columnName) ? null : row.getUDTValue(columnName);
+ if (paramValue != null) {
+ try {
+ PersistentBase udtObject = (PersistentBase) SpecificData.newInstance(Class.forName(schema.getFullName()), schema);
+ for (Schema.Field f : udtObject.getSchema().getFields()) {
+ DataType dType = ((UDTValue) paramValue).getType().getFieldType(f.name());
+ Object fieldValue = getValue((UDTValue) paramValue, dType, f.name(), f.schema());
+ udtObject.put(f.pos(), fieldValue);
+ }
+ paramValue = udtObject;
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Error occurred while populating data to " + schema.getFullName() + " : " + e.getMessage());
+ }
+ }
break;
case TUPLE:
paramValue = row.isNull(columnName) ? null : row.getTupleValue(columnName).toString();
@@ -234,35 +280,35 @@ class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer {
return paramValue;
}
-/* public Collection<Object> getFieldValues(Object o) {
- UDTValue udtValue = (UDTValue) o;
- UserType type = udtValue.getType();
-
- Collection<Object> values = new ArrayList<Object>(type.size());
-
- *//* for (UserType.Field field : type) {
- udtValue.
- ByteBuffer bytes = udtValue.getBytesUnsafe(field.getName());
- DataType value = field.getType();
- for(DataType type1 : value.getTypeArguments()) {
- type1.
- }
- values.add(value);
- }*//*
-
- return values;
- }*/
-
-
private Class getRelevantClassForCassandraDataType(String dataType) {
switch (dataType) {
- //// TODO: 7/25/17 support all the datatypes
case "ascii":
case "text":
case "varchar":
return String.class;
case "blob":
return ByteBuffer.class;
+ case "int":
+ return Integer.class;
+ case "double":
+ return Double.class;
+ case "bigint":
+ case "counter":
+ return Long.class;
+ case "decimal":
+ return BigDecimal.class;
+ case "float":
+ return Float.class;
+ case "boolean":
+ return Boolean.class;
+ case "inet":
+ return InetAddress.class;
+ case "varint":
+ return BigInteger.class;
+ case "uuid":
+ return UUID.class;
+ case "timestamp":
+ return Date.class;
default:
throw new RuntimeException("Invalid Cassandra DataType");
}
@@ -302,7 +348,7 @@ class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer {
K keyObject;
CassandraKey cassandraKey = mapping.getCassandraKey();
while (iterator.hasNext()) {
- Row row = iterator.next();
+ AbstractGettableData row = (AbstractGettableData) iterator.next();
obj = cassandraDataStore.newPersistent();
keyObject = cassandraDataStore.newKey();
populateValuesToPersistent(row, definitions, obj, fields);
@@ -310,7 +356,7 @@ class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer {
populateValuesToPersistent(row, definitions, (PersistentBase) keyObject, cassandraKey.getFieldNames());
} else {
Field key = mapping.getInlinedDefinedPartitionKey();
- keyObject = (K) getValue(row, definitions, key.getColumnName());
+ keyObject = (K) getValue(row, definitions.getType(key.getColumnName()), key.getColumnName(), null);
}
cassandraResult.addResultElement(keyObject, obj);
}
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/CassandraQueryFactory.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/CassandraQueryFactory.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/CassandraQueryFactory.java
index 10c8f68..184955c 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/CassandraQueryFactory.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/CassandraQueryFactory.java
@@ -16,28 +16,31 @@
*/
package org.apache.gora.cassandra.serializers;
+import com.datastax.driver.core.querybuilder.BuiltStatement;
import com.datastax.driver.core.querybuilder.Delete;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;
import com.datastax.driver.core.querybuilder.Update;
+import com.datastax.driver.mapping.annotations.UDT;
import org.apache.avro.Schema;
import org.apache.gora.cassandra.bean.CassandraKey;
import org.apache.gora.cassandra.bean.ClusterKeyField;
import org.apache.gora.cassandra.bean.Field;
import org.apache.gora.cassandra.bean.KeySpace;
import org.apache.gora.cassandra.bean.PartitionKeyField;
-import org.apache.gora.cassandra.persistent.CassandraNativePersistent;
import org.apache.gora.cassandra.query.CassandraQuery;
import org.apache.gora.cassandra.store.CassandraMapping;
+import org.apache.gora.cassandra.store.CassandraStore;
import org.apache.gora.query.Query;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import java.util.Set;
/**
* This class is used create Cassandra Queries.
@@ -95,14 +98,12 @@ class CassandraQueryFactory {
static String getCreateTableQuery(CassandraMapping mapping) {
StringBuilder stringBuffer = new StringBuilder();
stringBuffer.append("CREATE TABLE IF NOT EXISTS ").append(mapping.getKeySpace().getName()).append(".").append(mapping.getCoreName()).append(" (");
- boolean isCommaNeeded = false;
CassandraKey cassandraKey = mapping.getCassandraKey();
// appending Cassandra Persistent columns into db schema
- processFieldsForCreateTableQuery(mapping.getFieldList(), isCommaNeeded, stringBuffer);
+ processFieldsForCreateTableQuery(mapping.getFieldList(), false, stringBuffer);
if (cassandraKey != null) {
- isCommaNeeded = true;
- processFieldsForCreateTableQuery(cassandraKey.getFieldList(), isCommaNeeded, stringBuffer);
+ processFieldsForCreateTableQuery(cassandraKey.getFieldList(), true, stringBuffer);
List<PartitionKeyField> partitionKeys = cassandraKey.getPartitionKeyFields();
if (partitionKeys != null) {
stringBuffer.append(", PRIMARY KEY (");
@@ -253,17 +254,29 @@ class CassandraQueryFactory {
String[] objects = new String[fields.size()];
Arrays.fill(objects, "?");
Delete delete = QueryBuilder.delete().from(mapping.getKeySpace().getName(), mapping.getCoreName());
- Delete.Where query = null;
+ return processKeys(columnNames, delete);
+ }
+
+ private static String processKeys(String[] columnNames, BuiltStatement delete) {
+ BuiltStatement query = null;
boolean isWhereNeeded = true;
for (String columnName : columnNames) {
if (isWhereNeeded) {
- query = delete.where(QueryBuilder.eq(columnName, "?"));
+ if (delete instanceof Delete) {
+ query = ((Delete) delete).where(QueryBuilder.eq(columnName, "?"));
+ } else {
+ query = ((Select) delete).where(QueryBuilder.eq(columnName, "?"));
+ }
isWhereNeeded = false;
} else {
- query = query.and(QueryBuilder.eq(columnName, "?"));
+ if (delete instanceof Delete) {
+ query = ((Delete.Where) query).and(QueryBuilder.eq(columnName, "?"));
+ } else {
+ query = ((Select.Where) query).and(QueryBuilder.eq(columnName, "?"));
+ }
}
}
- return query.getQueryString();
+ return query != null ? query.getQueryString() : null;
}
/**
@@ -280,17 +293,7 @@ class CassandraQueryFactory {
select.allowFiltering();
}
String[] columnNames = getColumnNames(mapping, keyFields);
- Select.Where query = null;
- boolean isWhereNeeded = true;
- for (String columnName : columnNames) {
- if (isWhereNeeded) {
- query = select.where(QueryBuilder.eq(columnName, "?"));
- isWhereNeeded = false;
- } else {
- query = query.and(QueryBuilder.eq(columnName, "?"));
- }
- }
- return query.getQueryString();
+ return processKeys(columnNames, select);
}
/**
@@ -309,17 +312,7 @@ class CassandraQueryFactory {
select.allowFiltering();
}
String[] columnNames = getColumnNames(mapping, keyFields);
- Select.Where query = null;
- boolean isWhereNeeded = true;
- for (String columnName : columnNames) {
- if (isWhereNeeded) {
- query = select.where(QueryBuilder.eq(columnName, "?"));
- isWhereNeeded = false;
- } else {
- query = query.and(QueryBuilder.eq(columnName, "?"));
- }
- }
- return query.getQueryString();
+ return processKeys(columnNames, select);
}
/**
@@ -373,10 +366,6 @@ class CassandraQueryFactory {
* @return CQL Query
*/
static String getExecuteQuery(CassandraMapping mapping, Query cassandraQuery, List<Object> objects, String[] fields) {
- Object startKey = cassandraQuery.getStartKey();
- Object endKey = cassandraQuery.getEndKey();
- Object key = cassandraQuery.getKey();
- String primaryKey = null;
long limit = cassandraQuery.getLimit();
Select select = QueryBuilder.select(getColumnNames(mapping, Arrays.asList(fields))).from(mapping.getKeySpace().getName(), mapping.getCoreName());
if (limit > 0) {
@@ -385,7 +374,15 @@ class CassandraQueryFactory {
if (Boolean.parseBoolean(mapping.getProperty("allowFiltering"))) {
select.allowFiltering();
}
- Select.Where query = null;
+ return processQuery(cassandraQuery, select, mapping, objects);
+ }
+
+ private static String processQuery(Query cassandraQuery, BuiltStatement select, CassandraMapping mapping, List<Object> objects) {
+ String primaryKey = null;
+ BuiltStatement query = null;
+ Object startKey = cassandraQuery.getStartKey();
+ Object endKey = cassandraQuery.getEndKey();
+ Object key = cassandraQuery.getKey();
boolean isWhereNeeded = true;
if (key != null) {
if (mapping.getCassandraKey() != null) {
@@ -395,17 +392,35 @@ class CassandraQueryFactory {
String[] columnKeys = getColumnNames(mapping, cassandraKeys);
for (int i = 0; i < cassandraKeys.size(); i++) {
if (isWhereNeeded) {
- query = select.where(QueryBuilder.eq(columnKeys[i], "?"));
+ if (select instanceof Select) {
+ query = ((Select) select).where(QueryBuilder.eq(columnKeys[i], "?"));
+ } else if (select instanceof Delete) {
+ query = ((Delete) select).where(QueryBuilder.eq(columnKeys[i], "?"));
+ } else {
+ query = ((Update.Assignments) select).where(QueryBuilder.eq(columnKeys[i], "?"));
+ }
objects.add(cassandraValues.get(i));
isWhereNeeded = false;
} else {
- query = query.and(QueryBuilder.eq(columnKeys[i], "?"));
+ if (select instanceof Select) {
+ query = ((Select.Where) query).and(QueryBuilder.eq(columnKeys[i], "?"));
+ } else if (select instanceof Delete) {
+ query = ((Delete.Where) query).and(QueryBuilder.eq(columnKeys[i], "?"));
+ } else {
+ query = ((Update.Where) query).and(QueryBuilder.eq(columnKeys[i], "?"));
+ }
objects.add(cassandraValues.get(i));
}
}
} else {
primaryKey = getPKey(mapping.getFieldList());
- query = select.where(QueryBuilder.eq(primaryKey, "?"));
+ if (select instanceof Select) {
+ query = ((Select) select).where(QueryBuilder.eq(primaryKey, "?"));
+ } else if (select instanceof Delete) {
+ query = ((Delete) select).where(QueryBuilder.eq(primaryKey, "?"));
+ } else {
+ query = ((Update.Assignments) select).where(QueryBuilder.eq(primaryKey, "?"));
+ }
objects.add(key);
}
} else {
@@ -417,17 +432,44 @@ class CassandraQueryFactory {
String[] columnKeys = getColumnNames(mapping, cassandraKeys);
for (int i = 0; i < cassandraKeys.size(); i++) {
if (isWhereNeeded) {
- query = select.where(QueryBuilder.gte(columnKeys[i], "?"));
+ if (select instanceof Select) {
+ query = ((Select) select).where(QueryBuilder.gte(columnKeys[i], "?"));
+ } else if (select instanceof Delete) {
+ /*
+ According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+ */
+ throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+ } else {
+ query = ((Update.Assignments) select).where(QueryBuilder.gte(columnKeys[i], "?"));
+ }
objects.add(cassandraValues.get(i));
isWhereNeeded = false;
} else {
- query = query.and(QueryBuilder.gte(columnKeys[i], "?"));
+ if (select instanceof Select) {
+ query = ((Select.Where) query).and(QueryBuilder.gte(columnKeys[i], "?"));
+ } else if (select instanceof Delete) {
+ /*
+ According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+ */
+ throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+ } else {
+ query = ((Update.Where) query).and(QueryBuilder.gte(columnKeys[i], "?"));
+ }
objects.add(cassandraValues.get(i));
}
}
} else {
primaryKey = getPKey(mapping.getFieldList());
- query = select.where(QueryBuilder.gte(primaryKey, "?"));
+ if (select instanceof Select) {
+ query = ((Select) select).where(QueryBuilder.gte(primaryKey, "?"));
+ } else if (select instanceof Delete) {
+ /*
+ According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+ */
+ throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+ } else {
+ query = ((Update.Assignments) select).where(QueryBuilder.gte(primaryKey, "?"));
+ }
objects.add(startKey);
isWhereNeeded = false;
}
@@ -440,20 +482,56 @@ class CassandraQueryFactory {
String[] columnKeys = getColumnNames(mapping, cassandraKeys);
for (int i = 0; i < cassandraKeys.size(); i++) {
if (isWhereNeeded) {
- query = select.where(QueryBuilder.lte(columnKeys[i], "?"));
+ if (select instanceof Select) {
+ query = ((Select) select).where(QueryBuilder.lte(columnKeys[i], "?"));
+ } else if (select instanceof Delete) {
+ /*
+ According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+ */
+ throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+ } else {
+ query = ((Update.Assignments) select).where(QueryBuilder.lte(columnKeys[i], "?"));
+ }
objects.add(cassandraValues.get(i));
isWhereNeeded = false;
} else {
- query = query.and(QueryBuilder.lte(columnKeys[i], "?"));
+ if (select instanceof Select) {
+ query = ((Select.Where) query).and(QueryBuilder.lte(columnKeys[i], "?"));
+ } else if (select instanceof Delete) {
+ /*
+ According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+ */
+ throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+ } else {
+ query = ((Update.Where) query).and(QueryBuilder.lte(columnKeys[i], "?"));
+ }
objects.add(cassandraValues.get(i));
}
}
} else {
primaryKey = primaryKey != null ? primaryKey : getPKey(mapping.getFieldList());
if (isWhereNeeded) {
- query = select.where(QueryBuilder.lte(primaryKey, "?"));
+ if (select instanceof Select) {
+ query = ((Select) select).where(QueryBuilder.lte(primaryKey, "?"));
+ } else if (select instanceof Delete) {
+ /*
+ According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+ */
+ throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+ } else {
+ query = ((Update.Assignments) select).where(QueryBuilder.lte(primaryKey, "?"));
+ }
} else {
- query = query.and(QueryBuilder.lte(primaryKey, "?"));
+ if (select instanceof Select) {
+ query = ((Select.Where) query).and(QueryBuilder.lte(primaryKey, "?"));
+ } else if (select instanceof Delete) {
+ /*
+ According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+ */
+ throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+ } else {
+ query = ((Update.Where) query).and(QueryBuilder.lte(primaryKey, "?"));
+ }
}
objects.add(endKey);
}
@@ -462,7 +540,7 @@ class CassandraQueryFactory {
if (startKey == null && endKey == null && key == null) {
return select.getQueryString();
}
- return query.getQueryString();
+ return query != null ? query.getQueryString() : null;
}
private static String[] getColumnNames(CassandraMapping mapping, List<String> fields) {
@@ -509,94 +587,13 @@ class CassandraQueryFactory {
if (cassandraQuery.getFields() != null) {
columns = getColumnNames(mapping, Arrays.asList(cassandraQuery.getFields()));
}
- Object startKey = cassandraQuery.getStartKey();
- Object endKey = cassandraQuery.getEndKey();
- Object key = cassandraQuery.getKey();
- String primaryKey = null;
Delete delete;
if (columns != null) {
delete = QueryBuilder.delete(columns).from(mapping.getKeySpace().getName(), mapping.getCoreName());
} else {
delete = QueryBuilder.delete().from(mapping.getKeySpace().getName(), mapping.getCoreName());
}
- Delete.Where query = null;
- boolean isWhereNeeded = true;
- if (key != null) {
- if (mapping.getCassandraKey() != null) {
- ArrayList<String> cassandraKeys = new ArrayList<>();
- ArrayList<Object> cassandraValues = new ArrayList<>();
- AvroCassandraUtils.processKeys(mapping, key, cassandraKeys, cassandraValues);
- String[] columnKeys = getColumnNames(mapping, cassandraKeys);
- for (int i = 0; i < cassandraKeys.size(); i++) {
- if (isWhereNeeded) {
- query = delete.where(QueryBuilder.eq(columnKeys[i], "?"));
- objects.add(cassandraValues.get(i));
- isWhereNeeded = false;
- } else {
- query = query.and(QueryBuilder.eq(columnKeys[i], "?"));
- objects.add(cassandraValues.get(i));
- }
- }
- } else {
- primaryKey = getPKey(mapping.getFieldList());
- query = delete.where(QueryBuilder.eq(primaryKey, "?"));
- objects.add(key);
- }
- } else {
- if (startKey != null) {
- if (mapping.getCassandraKey() != null) {
- ArrayList<String> cassandraKeys = new ArrayList<>();
- ArrayList<Object> cassandraValues = new ArrayList<>();
- AvroCassandraUtils.processKeys(mapping, startKey, cassandraKeys, cassandraValues);
- String[] columnKeys = getColumnNames(mapping, cassandraKeys);
- for (int i = 0; i < cassandraKeys.size(); i++) {
- if (isWhereNeeded) {
- query = delete.where(QueryBuilder.gte(QueryBuilder.token(columnKeys[i]), QueryBuilder.token("?")));
- objects.add(cassandraValues.get(i));
- isWhereNeeded = false;
- } else {
- query = query.and(QueryBuilder.gte(QueryBuilder.token(columnKeys[i]), QueryBuilder.token("?")));
- objects.add(cassandraValues.get(i));
- }
- }
- } else {
- primaryKey = getPKey(mapping.getFieldList());
- query = delete.where(QueryBuilder.gte(QueryBuilder.token(primaryKey), QueryBuilder.token("?")));
- objects.add(startKey);
- isWhereNeeded = false;
- }
- }
- if (endKey != null) {
- if (mapping.getCassandraKey() != null) {
- ArrayList<String> cassandraKeys = new ArrayList<>();
- ArrayList<Object> cassandraValues = new ArrayList<>();
- AvroCassandraUtils.processKeys(mapping, endKey, cassandraKeys, cassandraValues);
- String[] columnKeys = getColumnNames(mapping, cassandraKeys);
- for (int i = 0; i < cassandraKeys.size(); i++) {
- if (isWhereNeeded) {
- query = delete.where(QueryBuilder.lte(QueryBuilder.token(columnKeys[i]), QueryBuilder.token("?")));
- objects.add(cassandraValues.get(i));
- isWhereNeeded = false;
- } else {
- query = query.and(QueryBuilder.lte(QueryBuilder.token(columnKeys[i]), QueryBuilder.token("?")));
- objects.add(cassandraValues.get(i));
- }
- }
- } else {
- primaryKey = primaryKey != null ? primaryKey : getPKey(mapping.getFieldList());
- if (isWhereNeeded) {
- query = delete.where(QueryBuilder.lte(QueryBuilder.token(primaryKey), QueryBuilder.token("?")));
- } else {
- query = query.and(QueryBuilder.lte(QueryBuilder.token(primaryKey), QueryBuilder.token("?")));
- }
- objects.add(endKey);
- }
- }
- }
- if (startKey == null && endKey == null && key == null) {
- return delete.getQueryString();
- }
- return query.getQueryString();
+ return processQuery(cassandraQuery, delete, mapping, objects);
}
/**
@@ -608,12 +605,12 @@ class CassandraQueryFactory {
* @param objects field Objects list
* @return CQL Query
*/
- static String getUpdateByQuery(CassandraMapping mapping, Query cassandraQuery, List<Object> objects) {
+ static String getUpdateByQuery(CassandraMapping mapping, Query cassandraQuery, List<Object> objects, Schema schema) {
Update update = QueryBuilder.update(mapping.getKeySpace().getName(), mapping.getCoreName());
Update.Assignments updateAssignments = null;
if (cassandraQuery instanceof CassandraQuery) {
String[] columnNames = getColumnNames(mapping, Arrays.asList(cassandraQuery.getFields()));
- if (CassandraNativePersistent.class.isAssignableFrom(mapping.getPersistentClass())) {
+ if (((CassandraStore) cassandraQuery.getDataStore()).getSerializationType().equalsIgnoreCase("NATIVE")) {
for (String column : columnNames) {
updateAssignments = update.with(QueryBuilder.set(column, "?"));
objects.add(((CassandraQuery) cassandraQuery).getUpdateFieldValue(mapping.getFieldFromColumnName(column).getFieldName()));
@@ -624,159 +621,33 @@ class CassandraQueryFactory {
Field field = mapping.getFieldFromColumnName(column);
Object value = ((CassandraQuery) cassandraQuery).getUpdateFieldValue(field.getFieldName());
try {
- Schema schema = (Schema) mapping.getPersistentClass().getField("SCHEMA$").get(null);
Schema schemaField = schema.getField(field.getFieldName()).schema();
objects.add(AvroCassandraUtils.getFieldValueFromAvroBean(schemaField, schemaField.getType(), value, field));
- } catch (IllegalAccessException | NoSuchFieldException e) {
- throw new RuntimeException("SCHEMA$ field can't accessible, Please recompile the Avro schema with goracompiler.");
} catch (NullPointerException e) {
throw new RuntimeException(field + " field couldn't find in the class " + mapping.getPersistentClass() + ".");
}
}
}
- }
- String primaryKey = null;
- Update.Where query = null;
- Object startKey = cassandraQuery.getStartKey();
- Object endKey = cassandraQuery.getEndKey();
- Object key = cassandraQuery.getKey();
- boolean isWhereNeeded = true;
- if (key != null) {
- if (mapping.getCassandraKey() != null) {
- ArrayList<String> cassandraKeys = new ArrayList<>();
- ArrayList<Object> cassandraValues = new ArrayList<>();
- AvroCassandraUtils.processKeys(mapping, key, cassandraKeys, cassandraValues);
- String[] columnKeys = getColumnNames(mapping, cassandraKeys);
- for (int i = 0; i < cassandraKeys.size(); i++) {
- if (isWhereNeeded) {
- query = updateAssignments.where(QueryBuilder.eq(columnKeys[i], "?"));
- objects.add(cassandraValues.get(i));
- isWhereNeeded = false;
- } else {
- query = query.and(QueryBuilder.eq(columnKeys[i], "?"));
- objects.add(cassandraValues.get(i));
- }
- }
- } else {
- primaryKey = getPKey(mapping.getFieldList());
- query = updateAssignments.where(QueryBuilder.eq(primaryKey, "?"));
- objects.add(key);
- }
} else {
- if (startKey != null) {
- if (mapping.getCassandraKey() != null) {
- ArrayList<String> cassandraKeys = new ArrayList<>();
- ArrayList<Object> cassandraValues = new ArrayList<>();
- AvroCassandraUtils.processKeys(mapping, startKey, cassandraKeys, cassandraValues);
- String[] columnKeys = getColumnNames(mapping, cassandraKeys);
- for (int i = 0; i < cassandraKeys.size(); i++) {
- if (isWhereNeeded) {
- query = updateAssignments.where(QueryBuilder.gte(columnKeys[i], "?"));
- objects.add(cassandraValues.get(i));
- isWhereNeeded = false;
- } else {
- query = query.and(QueryBuilder.gte(columnKeys[i], "?"));
- objects.add(cassandraValues.get(i));
- }
- }
- } else {
- primaryKey = getPKey(mapping.getFieldList());
- query = updateAssignments.where(QueryBuilder.gte(primaryKey, "?"));
- objects.add(startKey);
- isWhereNeeded = false;
- }
- }
- if (endKey != null) {
- if (mapping.getCassandraKey() != null) {
- ArrayList<String> cassandraKeys = new ArrayList<>();
- ArrayList<Object> cassandraValues = new ArrayList<>();
- AvroCassandraUtils.processKeys(mapping, endKey, cassandraKeys, cassandraValues);
- String[] columnKeys = getColumnNames(mapping, cassandraKeys);
- for (int i = 0; i < cassandraKeys.size(); i++) {
- if (isWhereNeeded) {
- query = updateAssignments.where(QueryBuilder.lte(columnKeys[i], "?"));
- objects.add(cassandraValues.get(i));
- isWhereNeeded = false;
- } else {
- query = query.and(QueryBuilder.lte(columnKeys[i], "?"));
- objects.add(cassandraValues.get(i));
- }
- }
- } else {
- primaryKey = primaryKey != null ? primaryKey : getPKey(mapping.getFieldList());
- if (isWhereNeeded) {
- query = updateAssignments.where(QueryBuilder.lte(primaryKey, "?"));
- } else {
- query = query.and(QueryBuilder.lte(primaryKey, "?"));
- }
- objects.add(endKey);
- }
- }
- }
- if (startKey == null && endKey == null && key == null) {
- return updateAssignments.getQueryString();
+ throw new RuntimeException("Please use Cassandra Query object to invoke, UpdateByQuery method.");
}
- return query.getQueryString();
- }
- /**
- * This method returns create Type CQL query to create user define types.
- * refer : http://docs.datastax.com/en/cql/3.1/cql/cql_reference/cqlRefcreateType.html
- *
- * @param fieldSchema avroSchema {@link Schema}
- * @param mapping Cassandra mapping {@link CassandraMapping}
- * @return CQL Query
- */
- static String getCreateUDTType(Schema fieldSchema, CassandraMapping mapping, Set<String> udtQueryStack) {
- StringBuilder stringBuffer = new StringBuilder();
- if (fieldSchema.getType().equals(Schema.Type.UNION)) {
- for (Schema fieldTypeSchema : fieldSchema.getTypes()) {
- if (fieldTypeSchema.getType().equals(Schema.Type.RECORD)) {
- fieldSchema = fieldTypeSchema;
- break;
- }
- }
- }
- stringBuffer.append("CREATE TYPE IF NOT EXISTS ").append(mapping.getKeySpace().getName()).append(".").append(fieldSchema.getName()).append(" (");
- processRecord(fieldSchema, stringBuffer, mapping, udtQueryStack);
- stringBuffer.append(")");
- return stringBuffer.toString();
+ return processQuery(cassandraQuery, updateAssignments, mapping, objects);
}
- private static void processRecord(Schema recordSchema, StringBuilder stringBuilder, CassandraMapping mapping, Set<String> udtQueryStack) {
- boolean isCommaNeeded = false;
- for (Schema.Field field : recordSchema.getFields()) {
- if (isCommaNeeded) {
- stringBuilder.append(", ");
- }
- String fieldName = field.name();
- stringBuilder.append(fieldName).append(" ");
- try {
- populateFieldsToQuery(field.schema(), stringBuilder, mapping, udtQueryStack);
- isCommaNeeded = true;
- } catch (Exception e) {
- int i = stringBuilder.indexOf(fieldName);
- if (i != -1) {
- stringBuilder.delete(i, i + fieldName.length());
- isCommaNeeded = false;
- }
- }
- }
- }
-
- private static void populateFieldsToQuery(Schema schema, StringBuilder builder, CassandraMapping mapping, Set<String> udtQueryStack) throws Exception {
+ private static void populateFieldsToQuery(Schema schema, StringBuilder builder) throws Exception {
switch (schema.getType()) {
case INT:
builder.append("int");
break;
case MAP:
builder.append("map<text,");
- populateFieldsToQuery(schema.getValueType(), builder, mapping, udtQueryStack);
+ populateFieldsToQuery(schema.getValueType(), builder);
builder.append(">");
break;
case ARRAY:
builder.append("list<");
- populateFieldsToQuery(schema.getElementType(), builder, mapping, udtQueryStack);
+ populateFieldsToQuery(schema.getElementType(), builder);
builder.append(">");
break;
case LONG:
@@ -796,8 +667,6 @@ class CassandraQueryFactory {
break;
case RECORD:
builder.append("frozen<").append(schema.getName()).append(">");
- String query = getCreateUDTType(schema, mapping, udtQueryStack);
- udtQueryStack.add(query);
break;
case STRING:
case FIXED:
@@ -810,15 +679,13 @@ class CassandraQueryFactory {
String recordName = unionElementSchema.getName();
if (!builder.toString().contains(recordName)) {
builder.append("frozen<").append(recordName).append(">");
- query = getCreateUDTType(unionElementSchema, mapping, udtQueryStack);
- udtQueryStack.add(query);
} else {
LOG.warn("Same Field Type can't be mapped recursively. This is not supported with Cassandra UDT types, Please use byte dataType for recursive mapping.");
throw new Exception("Same Field Type has mapped recursively");
}
break;
} else if (!unionElementSchema.getType().equals(Schema.Type.NULL)) {
- populateFieldsToQuery(unionElementSchema, builder, mapping, udtQueryStack);
+ populateFieldsToQuery(unionElementSchema, builder);
break;
}
}
@@ -826,4 +693,127 @@ class CassandraQueryFactory {
}
}
+ static void processRecord(Schema recordSchema, StringBuilder stringBuilder) {
+ boolean isCommaNeeded = false;
+ for (Schema.Field field : recordSchema.getFields()) {
+ if (isCommaNeeded) {
+ stringBuilder.append(", ");
+ }
+ String fieldName = field.name();
+ stringBuilder.append(fieldName).append(" ");
+ try {
+ populateFieldsToQuery(field.schema(), stringBuilder);
+ isCommaNeeded = true;
+ } catch (Exception e) {
+ int i = stringBuilder.indexOf(fieldName);
+ if (i != -1) {
+ stringBuilder.delete(i, i + fieldName.length());
+ isCommaNeeded = false;
+ }
+ }
+ }
+ }
+
+ static String getCreateUDTTypeForNative(CassandraMapping mapping, Class persistentClass, String udtType, String fieldName) throws NoSuchFieldException {
+ StringBuilder stringBuffer = new StringBuilder();
+ Class udtClass = persistentClass.getDeclaredField(fieldName).getType();
+ UDT annotation = (UDT) udtClass.getAnnotation(UDT.class);
+ if (annotation != null) {
+ stringBuffer.append("CREATE TYPE IF NOT EXISTS ").append(mapping.getKeySpace().getName()).append(".").append(udtType).append(" (");
+ boolean isCommaNeeded = false;
+ for (java.lang.reflect.Field udtField : udtClass.getDeclaredFields()) {
+ com.datastax.driver.mapping.annotations.Field fieldAnnotation = udtField.getDeclaredAnnotation(com.datastax.driver.mapping.annotations.Field.class);
+ if (fieldAnnotation != null) {
+ if (isCommaNeeded) {
+ stringBuffer.append(", ");
+ }
+ if(!fieldAnnotation.name().isEmpty()) {
+ stringBuffer.append(fieldAnnotation.name()).append(" ");
+ } else {
+ stringBuffer.append(udtField.getName()).append(" ");
+ }
+ stringBuffer.append(dataType(udtField, null));
+ isCommaNeeded = true;
+ }
+ }
+ stringBuffer.append(")");
+ } else {
+ throw new RuntimeException("");
+ }
+ return stringBuffer.toString();
+ }
+
+ static String getCreateUDTTypeForAvro(CassandraMapping mapping, String udtType, Schema fieldSchema) {
+ StringBuilder stringBuffer = new StringBuilder();
+ stringBuffer.append("CREATE TYPE IF NOT EXISTS ").append(mapping.getKeySpace().getName()).append(".").append(udtType).append(" (");
+ CassandraQueryFactory.processRecord(fieldSchema, stringBuffer);
+ stringBuffer.append(")");
+ return stringBuffer.toString();
+ }
+
+ private static String dataType(java.lang.reflect.Field field, Type fieldType) {
+ String type;
+ if (field != null) {
+ type = field.getType().getName();
+ } else {
+ type = fieldType.getTypeName();
+ }
+ String dataType;
+ String s = type;
+ if (s.equals("java.lang.String") || s.equals("java.lang.CharSequence")) {
+ dataType = "text";
+ } else if (s.equals("int") || s.equals("java.lang.Integer")) {
+ dataType = "int";
+ } else if (s.equals("double") || s.equals("java.lang.Double")) {
+ dataType = "double";
+ } else if (s.equals("float") || s.equals("java.lang.Float")) {
+ dataType = "float";
+ } else if (s.equals("boolean") || s.equals("java.lang.Boolean")) {
+ dataType = "boolean";
+ } else if (s.equals("java.util.UUID")) {
+ dataType = "uuid";
+ } else if (s.equals("java.lang.Long")) {
+ dataType = "bigint";
+ }else if (s.equals("java.math.BigDecimal")) {
+ dataType = "decimal";
+ }else if (s.equals("java.net.InetAddress")) {
+ dataType = "inet";
+ }else if (s.equals("java.math.BigInteger")) {
+ dataType = "varint";
+ } else if (s.equals("java.nio.ByteBuffer")) {
+ dataType = "blob";
+ }else if (s.contains("Map")) {
+ ParameterizedType mapType;
+ if (field != null) {
+ mapType = (ParameterizedType) field.getGenericType();
+ } else {
+ mapType = (ParameterizedType) fieldType;
+ }
+ Type value1 = mapType.getActualTypeArguments()[0];
+ Type value2 = mapType.getActualTypeArguments()[1];
+ dataType = "map<" +dataType(null, value1) + "," + dataType(null, value2) + ">";
+ } else if (s.contains("List")) {
+ ParameterizedType listType;
+ if (field != null) {
+ listType = (ParameterizedType) field.getGenericType();
+ } else {
+ listType = (ParameterizedType) fieldType;
+ }
+ Type value = listType.getActualTypeArguments()[0];
+ dataType = "list<" + dataType(null, value) + ">";
+ } else if (s.contains("Set")) {
+ ParameterizedType setType;
+ if (field != null) {
+ setType = (ParameterizedType) field.getGenericType();
+ } else {
+ setType = (ParameterizedType) fieldType;
+ }
+ Type value = setType.getActualTypeArguments()[0];
+ dataType = "set<" + dataType(null, value) + ">";
+ } else {
+ throw new RuntimeException("Unsupported Cassandra DataType");
+ }
+ return dataType;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/CassandraSerializer.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/CassandraSerializer.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/CassandraSerializer.java
index 17e0568..651dfdb 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/CassandraSerializer.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/CassandraSerializer.java
@@ -26,6 +26,7 @@ import org.apache.gora.cassandra.store.CassandraClient;
import org.apache.gora.cassandra.store.CassandraMapping;
import org.apache.gora.cassandra.store.CassandraStore;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.query.Result;
import org.apache.gora.store.DataStore;
@@ -33,10 +34,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
-import java.util.LinkedHashSet;
+import java.util.HashMap;
import java.util.List;
import java.util.Locale;
-import java.util.Set;
+import java.util.Map;
/**
* This is the abstract Cassandra Serializer class.
@@ -47,14 +48,23 @@ public abstract class CassandraSerializer<K, T extends Persistent> {
protected Class<T> persistentClass;
+ private Map<String, String> userDefineTypeMaps;
+
protected CassandraMapping mapping;
+ private Schema persistentSchema;
CassandraClient client;
- CassandraSerializer(CassandraClient cc, Class<K> keyClass, Class<T> persistantClass, CassandraMapping mapping) {
+ CassandraSerializer(CassandraClient cc, Class<K> keyClass, Class<T> persistantClass, CassandraMapping mapping, Schema schema) {
this.keyClass = keyClass;
this.persistentClass = persistantClass;
this.client = cc;
this.mapping = mapping;
+ persistentSchema = schema;
+ try {
+ analyzePersistent();
+ } catch (Exception e) {
+ throw new RuntimeException("Error occurred while analyzing the persistent class, :" + e.getMessage());
+ }
}
/**
@@ -68,24 +78,59 @@ public abstract class CassandraSerializer<K, T extends Persistent> {
* @param <T> persistent class
* @return Serializer
*/
- public static <K, T extends Persistent> CassandraSerializer getSerializer(CassandraClient cc, String type, final DataStore<K, T> dataStore, CassandraMapping mapping) {
+ public static <K, T extends Persistent> CassandraSerializer getSerializer(CassandraClient cc, String type, final DataStore<K, T> dataStore, CassandraMapping mapping, Schema schema) {
CassandraStore.SerializerType serType = type.isEmpty() ? CassandraStore.SerializerType.NATIVE : CassandraStore.SerializerType.valueOf(type.toUpperCase(Locale.ENGLISH));
CassandraSerializer serializer;
switch (serType) {
case AVRO:
- serializer = new AvroSerializer(cc, dataStore, mapping);
+ serializer = new AvroSerializer(cc, dataStore, mapping, schema);
break;
case NATIVE:
default:
- serializer = new NativeSerializer(cc, dataStore.getKeyClass(), dataStore.getPersistentClass(), mapping);
+ serializer = new NativeSerializer(cc, dataStore.getKeyClass(), dataStore.getPersistentClass(), mapping, schema);
}
return serializer;
}
+ private void analyzePersistent() throws Exception {
+ userDefineTypeMaps = new HashMap<>();
+ for (Field field : mapping.getFieldList()) {
+ String fieldType = field.getType();
+ if (fieldType.contains("frozen")) {
+ String udtType = fieldType.substring(fieldType.indexOf("<") + 1, fieldType.indexOf(">"));
+ if (this instanceof AvroSerializer) {
+ if (PersistentBase.class.isAssignableFrom(persistentClass)) {
+ Schema fieldSchema = persistentSchema.getField(field.getFieldName()).schema();
+ if (fieldSchema.getType().equals(Schema.Type.UNION)) {
+ for (Schema currentSchema : fieldSchema.getTypes()) {
+ if (currentSchema.getType().equals(Schema.Type.RECORD)) {
+ fieldSchema = currentSchema;
+ break;
+ }
+ }
+ }
+ String createQuery = CassandraQueryFactory.getCreateUDTTypeForAvro(mapping, udtType, fieldSchema);
+ userDefineTypeMaps.put(udtType, createQuery);
+ } else {
+ throw new RuntimeException("Unsupported Class for User Define Types, Please use PersistentBase class. field : " + udtType);
+ }
+ } else {
+ String createQuery = CassandraQueryFactory.getCreateUDTTypeForNative(mapping, persistentClass, udtType, field.getFieldName());
+ userDefineTypeMaps.put(udtType, createQuery);
+ }
+ }
+ }
+
+ }
+
+
public void createSchema() {
LOG.debug("creating Cassandra keyspace {}", mapping.getKeySpace().getName());
this.client.getSession().execute(CassandraQueryFactory.getCreateKeySpaceQuery(mapping));
- processUDTSchemas(); //TODO complete functionality
+ for (Map.Entry udtType : userDefineTypeMaps.entrySet()) {
+ LOG.debug("creating Cassandra User Define Type {}", udtType.getKey());
+ this.client.getSession().execute((String) udtType.getValue());
+ }
LOG.debug("creating Cassandra column family / table {}", mapping.getCoreName());
this.client.getSession().execute(CassandraQueryFactory.getCreateTableQuery(mapping));
}
@@ -116,32 +161,6 @@ public abstract class CassandraSerializer<K, T extends Persistent> {
}
}
- private void processUDTSchemas() {
- Set<String> schemaStack = new LinkedHashSet<>();
- for (Field field : mapping.getFieldList()) {
- if (field.getType().contains("frozen")) {
- try {
- Schema schema = (Schema) mapping.getPersistentClass().getField("SCHEMA$").get(null);
- Schema schemaField = schema.getField(field.getFieldName()).schema();
- String cqlQuery = CassandraQueryFactory.getCreateUDTType(schemaField, mapping, schemaStack);
- schemaStack.add(cqlQuery);
- } catch (IllegalAccessException | NoSuchFieldException e) {
- throw new RuntimeException("SCHEMA$ field can't accessible, Please recompile the Avro schema with goracompiler.");
- } catch (NullPointerException e) {
- throw new RuntimeException(field + " field couldn't find in the class " + mapping.getPersistentClass() + ".");
- }
- }
- }
- createUserDefineTypes(schemaStack);
-
- }
-
- private void createUserDefineTypes(Set<String> queries) {
- for (String cqlQuery : queries) {
- this.client.getSession().execute(cqlQuery);
- }
- }
-
protected String[] getFields() {
List<String> fields = new ArrayList<>();
for (Field field : mapping.getFieldList()) {
@@ -162,7 +181,7 @@ public abstract class CassandraSerializer<K, T extends Persistent> {
public boolean updateByQuery(Query query) {
List<Object> objectArrayList = new ArrayList<>();
- String cqlQuery = CassandraQueryFactory.getUpdateByQuery(mapping, query, objectArrayList);
+ String cqlQuery = CassandraQueryFactory.getUpdateByQuery(mapping, query, objectArrayList, persistentSchema);
ResultSet results;
if (objectArrayList.size() == 0) {
results = client.getSession().execute(cqlQuery);
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/NativeSerializer.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/NativeSerializer.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/NativeSerializer.java
index f8bb066..abff409 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/NativeSerializer.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/NativeSerializer.java
@@ -21,9 +21,9 @@ import com.datastax.driver.core.ResultSet;
import com.datastax.driver.mapping.Mapper;
import com.datastax.driver.mapping.MappingManager;
import com.datastax.driver.mapping.Result;
+import org.apache.avro.Schema;
import org.apache.commons.lang.ArrayUtils;
import org.apache.gora.cassandra.bean.Field;
-import org.apache.gora.cassandra.persistent.CassandraNativePersistent;
import org.apache.gora.cassandra.query.CassandraResultSet;
import org.apache.gora.cassandra.store.CassandraClient;
import org.apache.gora.cassandra.store.CassandraMapping;
@@ -41,14 +41,14 @@ import java.util.List;
/**
* This Class contains the operation relates to Native Serialization.
*/
-class NativeSerializer<K, T extends CassandraNativePersistent> extends CassandraSerializer {
+class NativeSerializer<K, T extends Persistent> extends CassandraSerializer {
private static final Logger LOG = LoggerFactory.getLogger(NativeSerializer.class);
private Mapper<T> mapper;
- NativeSerializer(CassandraClient cassandraClient, Class<K> keyClass, Class<T> persistentClass, CassandraMapping mapping) {
- super(cassandraClient, keyClass, persistentClass, mapping);
+ NativeSerializer(CassandraClient cassandraClient, Class<K> keyClass, Class<T> persistentClass, CassandraMapping mapping, Schema schema) {
+ super(cassandraClient, keyClass, persistentClass, mapping, schema);
this.createSchema();
MappingManager mappingManager = new MappingManager(cassandraClient.getSession());
mapper = mappingManager.mapper(persistentClass);
@@ -131,7 +131,7 @@ class NativeSerializer<K, T extends CassandraNativePersistent> extends Cassandra
break;
}
}
- K key = null;
+ K key;
Method keyMethod = null;
try {
for (Method method : this.persistentClass.getMethods()) {
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java
index ac46a30..807a99d 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java
@@ -51,6 +51,7 @@ public class CassandraMapping {
/**
* This method returns the KeySpace in the mapping file,
+ *
* @return Key space {@link KeySpace}
*/
public KeySpace getKeySpace() {
@@ -59,6 +60,7 @@ public class CassandraMapping {
/**
* This method set the KeySpace in the Cassandra mapping.
+ *
* @param keySpace Key space {@link KeySpace}
*/
void setKeySpace(KeySpace keySpace) {
@@ -67,6 +69,7 @@ public class CassandraMapping {
/**
* Thi method returns only the fields which belongs only for the Persistent Object.
+ *
* @return List of Fields
*/
public List<Field> getFieldList() {
@@ -75,6 +78,7 @@ public class CassandraMapping {
/**
* This method returns the Persistent Object's Field from the mapping, according to the FieldName.
+ *
* @param fieldName Field Name
* @return Field {@link Field}
*/
@@ -89,6 +93,7 @@ public class CassandraMapping {
/**
* This method returns the Persistent Object's Field from the mapping, according to the ColumnName.
+ *
* @param columnName Column Name
* @return Field {@link Field}
*/
@@ -103,6 +108,7 @@ public class CassandraMapping {
/**
* This method returns the Field Names
+ *
* @return array of Field Names
*/
public String[] getFieldNames() {
@@ -116,6 +122,7 @@ public class CassandraMapping {
/**
* This method returns
+ *
* @return
*/
public String[] getAllFieldsIncludingKeys() {
@@ -133,7 +140,6 @@ public class CassandraMapping {
}
/**
- *
* @return
*/
public String[] getAllKeys() {
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraMappingBuilder.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraMappingBuilder.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraMappingBuilder.java
index f151458..1d787de 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraMappingBuilder.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraMappingBuilder.java
@@ -242,6 +242,9 @@ class CassandraMappingBuilder<K, T extends Persistent> {
attributeValue = attributeValue.replace("udt(", "frozen(");
}
fieldKey.setType(attributeValue.replace("(", "<").replace(")", ">"));
+ if (fieldKey.getType().equalsIgnoreCase("udt")) {
+ throw new RuntimeException("Invalid udt type, Please enter dataType for udt with a unique name for particular user define data type, like udt(metadata).");
+ }
break;
default:
fieldKey.addProperty(attributeName, attributeValue);
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
index c481610..5e31fcc 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
@@ -17,12 +17,12 @@
package org.apache.gora.cassandra.store;
-import org.apache.avro.data.RecordBuilder;
-import org.apache.gora.cassandra.persistent.CassandraNativePersistent;
+import org.apache.avro.Schema;
import org.apache.gora.cassandra.query.CassandraQuery;
import org.apache.gora.cassandra.serializers.CassandraSerializer;
import org.apache.gora.persistency.BeanFactory;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.BeanFactoryImpl;
import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.PartitionQuery;
import org.apache.gora.query.Query;
@@ -54,16 +54,23 @@ public class CassandraStore<K, T extends Persistent> implements DataStore<K, T>
private Class<K> keyClass;
+ private Schema persistentSchema;
+
private Class<T> persistentClass;
private CassandraMapping mapping;
private CassandraSerializer cassandraSerializer;
+ private String serializationType;
public CassandraStore() {
super();
}
+ public String getSerializationType() {
+ return serializationType;
+ }
+
/**
* In initializing the cassandra datastore, read the mapping file, creates the basic connection to cassandra cluster,
* according to the gora properties
@@ -77,12 +84,21 @@ public class CassandraStore<K, T extends Persistent> implements DataStore<K, T>
try {
this.keyClass = keyClass;
this.persistentClass = persistentClass;
+ if (this.beanFactory == null) {
+ this.beanFactory = new BeanFactoryImpl<>(keyClass, persistentClass);
+ }
+ if (PersistentBase.class.isAssignableFrom(persistentClass)) {
+ persistentSchema = ((PersistentBase) this.beanFactory.getCachedPersistent()).getSchema();
+ } else {
+ persistentSchema = null;
+ }
String mappingFile = DataStoreFactory.getMappingFile(properties, this, DEFAULT_MAPPING_FILE);
+ serializationType = properties.getProperty(CassandraStoreParameters.CASSANDRA_SERIALIZATION_TYPE);
CassandraMappingBuilder mappingBuilder = new CassandraMappingBuilder(this);
mapping = mappingBuilder.readMapping(mappingFile);
CassandraClient cassandraClient = new CassandraClient();
cassandraClient.initialize(properties, mapping);
- cassandraSerializer = CassandraSerializer.getSerializer(cassandraClient, properties.getProperty(CassandraStoreParameters.CASSANDRA_SERIALIZATION_TYPE), this, mapping);
+ cassandraSerializer = CassandraSerializer.getSerializer(cassandraClient, serializationType, this, mapping, persistentSchema);
} catch (Exception e) {
throw new RuntimeException("Error while initializing Cassandra store: " + e.getMessage(), e);
}
@@ -100,7 +116,6 @@ public class CassandraStore<K, T extends Persistent> implements DataStore<K, T>
* This is a setter method to set the class of persistent objects.
*
* @param persistentClass class of persistent objects
- * {@link CassandraNativePersistent}
* {@link org.apache.gora.persistency.Persistent}
*/
@Override
@@ -153,9 +168,6 @@ public class CassandraStore<K, T extends Persistent> implements DataStore<K, T>
try {
if (beanFactory != null) {
return this.beanFactory.newPersistent();
- } else if (PersistentBase.class.isAssignableFrom(persistentClass)) {
- RecordBuilder builder = (RecordBuilder) persistentClass.getMethod("newBuilder").invoke(null);
- return (T) RecordBuilder.class.getMethod("build").invoke(builder);
} else {
return persistentClass.newInstance();
}
http://git-wip-us.apache.org/repos/asf/gora/blob/a9a3ad49/gora-cassandra-cql/src/test/conf/avroUDT/gora-cassandra-mapping.xml
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/test/conf/avroUDT/gora-cassandra-mapping.xml b/gora-cassandra-cql/src/test/conf/avroUDT/gora-cassandra-mapping.xml
new file mode 100644
index 0000000..9dc3c0b
--- /dev/null
+++ b/gora-cassandra-cql/src/test/conf/avroUDT/gora-cassandra-mapping.xml
@@ -0,0 +1,50 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+ 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.
+-->
+
+<gora-otd>
+
+ <keyspace name="avroKeySpace" durableWrite="false">
+ <placementStrategy name="SimpleStrategy" replication_factor="1"/>
+ </keyspace>
+
+ <class name="org.apache.gora.examples.generated.WebPage" keyClass="java.lang.String" table="WebPage"
+ allowFiltering="true"
+ keyspace="avroKeySpace">
+ <field name="url" column="url" type="ascii"/>
+ <field name="content" column="content" type="blob"/>
+ <field name="parsedContent" column="parsedContent" type="list(ascii)"/>
+ <field name="outlinks" column="outlinks" type="map(text,text)"/>
+ <field name="headers" column="headers" type="map(text,text)"/>
+ <field name="byteData" column="byteData" type="map(text,blob)"/>
+ <field name="metadata" column="metadata" type="udt(metadata)"/>
+ <field name="stringData" column="stringData" type="map(text,ascii)"/>
+ </class>
+
+ <class name="org.apache.gora.examples.generated.Employee" keyClass="java.lang.String" keyspace="avroKeySpace"
+ allowFiltering="true"
+ table="Employee" compactStorage="true">
+ <field name="name" column="name" type="text"/>
+ <field name="dateOfBirth" column="dob" type="bigint"/>
+ <field name="ssn" column="ssn" type="text"/>
+ <field name="salary" column="salary" type="int"/>
+ <field name="boss" column="boss" type="blob"/>
+ <field name="webpage" column="webpage" type="blob"/>
+ </class>
+
+</gora-otd>