http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/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 deleted file mode 100644 index b831c97..0000000 --- a/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/ComplexTypes.java +++ /dev/null @@ -1,199 +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 - * <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; -import java.util.Map; -import java.util.Set; -import java.util.UUID; - -/** - * Sample class for test native cassandra persistent. - */ -@Table(keyspace = "nativeTestKeySpace", name = "documents", - readConsistency = "QUORUM", - writeConsistency = "QUORUM", - caseSensitiveKeyspace = false, - caseSensitiveTable = true) -public class ComplexTypes implements Persistent { - - @Column - private List<String> listDataType; - @Column - private Map<String, String> mapDataType; - @Column - private String[] stringArrayDataType; - @Column - private int[] intArrayDataType; - @Column - private Set<String> setDataType; - @PartitionKey - @Column - private String id; - @Column - private List<UUID> listUUIDDataType; - - public ComplexTypes(String id) { - this.id = id; - } - - public ComplexTypes() { - } - - public List<UUID> getListUUIDDataType() { - return listUUIDDataType; - } - - public void setListUUIDDataType(List<UUID> listUUIDDataType) { - this.listUUIDDataType = listUUIDDataType; - } - - public List<String> getListDataType() { - return listDataType; - } - - public void setListDataType(List<String> listDataType) { - this.listDataType = listDataType; - } - - public Map<String, String> getMapDataType() { - return mapDataType; - } - - public void setMapDataType(Map<String, String> mapDataType) { - this.mapDataType = mapDataType; - } - - public String[] getStringArrayDataType() { - return stringArrayDataType; - } - - public void setStringArrayDataType(String[] stringArrayDataType) { - this.stringArrayDataType = stringArrayDataType; - } - - public int[] getIntArrayDataType() { - return intArrayDataType; - } - - public void setIntArrayDataType(int[] intArrayDataType) { - this.intArrayDataType = intArrayDataType; - } - - public Set<String> getSetDataType() { - return setDataType; - } - - public void setSetDataType(Set<String> setDataType) { - this.setDataType = setDataType; - } - - public String getId() { - return id; - } - - 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/89683c74/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 deleted file mode 100644 index 59bed36..0000000 --- a/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/Customer.java +++ /dev/null @@ -1,75 +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 - * <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.UUID; - -@UDT(name = "customer", keyspace = "nativeTestKeySpace") -public class 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 Customer() { - - } - - 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/89683c74/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 deleted file mode 100644 index 77961e4..0000000 --- a/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/Document.java +++ /dev/null @@ -1,136 +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 - * <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/89683c74/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 deleted file mode 100644 index 74b425e..0000000 --- a/gora-cassandra-cql/src/examples/java/org/apache/gora/cassandra/example/generated/nativeSerialization/User.java +++ /dev/null @@ -1,159 +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.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.Date; -import java.util.List; -import java.util.UUID; - -/** - * Sample class for native cassandra persistent example. - */ -@Table(keyspace = "nativeTestKeySpace", name = "users", - readConsistency = "QUORUM", - writeConsistency = "QUORUM", - caseSensitiveKeyspace = false, - caseSensitiveTable = false) -public class User implements Persistent { - @PartitionKey - @Column(name = "user_id") - private UUID userId; - @Column(name = "name") - private String name; - @Column(name = "dob") - private Date dateOfBirth; - - public User() { - - } - - public User(UUID userId, String name, Date dateOfBirth) { - this.userId = userId; - this.name = name; - this.dateOfBirth = dateOfBirth; - } - - public UUID getUserId() { - return userId; - } - - public void setUserId(UUID userId) { - this.userId = userId; - } - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - public Date getDateOfBirth() { - return dateOfBirth; - } - - 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/89683c74/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/CassandraKey.java ---------------------------------------------------------------------- diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/CassandraKey.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/CassandraKey.java deleted file mode 100644 index d3f3b1d..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/CassandraKey.java +++ /dev/null @@ -1,91 +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.bean; - -import java.util.ArrayList; -import java.util.List; - -/** - * This Class represents the Cassandra Key. - */ -public class CassandraKey { - - private String name; - - private List<ClusterKeyField> clusterKeyFields; - - private List<PartitionKeyField> partitionKeyFields; - - private List<Field> fieldList; - - public CassandraKey(String name) { - this.name = name; - this.fieldList = new ArrayList<>(); - this.partitionKeyFields = new ArrayList<>(); - } - - public String getName() { - return this.name; - } - - public List<ClusterKeyField> getClusterKeyFields() { - return this.clusterKeyFields; - } - - public List<PartitionKeyField> getPartitionKeyFields() { - return this.partitionKeyFields; - } - - public void addPartitionKeyField(PartitionKeyField partitionKeyField) { - this.partitionKeyFields.add(partitionKeyField); - if (partitionKeyField.isComposite()) { - this.fieldList.addAll(partitionKeyField.getFields()); - } else { - this.fieldList.add(partitionKeyField); - } - } - - public void addClusterKeyField(ClusterKeyField clusterKeyField) { - if (this.clusterKeyFields == null) { - this.clusterKeyFields = new ArrayList<>(); - } - this.clusterKeyFields.add(clusterKeyField); - } - - public String[] getFieldNames() { - List<String> fieldNames = new ArrayList<>(fieldList.size()); - for (Field field : fieldList) { - fieldNames.add(field.getFieldName()); - } - String[] fieldNameArray = new String[fieldNames.size()]; - return fieldNames.toArray(fieldNameArray); - } - - public Field getFieldFromFieldName(String fieldName) { - for (Field field1 : fieldList) { - if (field1.getFieldName().equalsIgnoreCase(fieldName)) { - return field1; - } - } - return null; - } - - public List<Field> getFieldList() { - return this.fieldList; - } - -} http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/ClusterKeyField.java ---------------------------------------------------------------------- diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/ClusterKeyField.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/ClusterKeyField.java deleted file mode 100644 index 7e23c75..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/ClusterKeyField.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.bean; - -/** - * This class represents Cassandra Clustering Key. - */ -public class ClusterKeyField { - - private String columnName; - - private Order order; - - public ClusterKeyField() { - - } - - public String getColumnName() { - return columnName; - } - - public void setColumnName(String columnName) { - this.columnName = columnName; - } - - public Order getOrder() { - return order; - } - - public void setOrder(Order order) { - this.order = order; - } - - public enum Order { - DESC, - ASC, - } - - -} http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/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 deleted file mode 100644 index b5015a1..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/Field.java +++ /dev/null @@ -1,70 +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.bean; - -import java.util.HashMap; -import java.util.Map; - -/** - * This Class represents the Cassandra Column. - */ -public class Field { - - private String fieldName; - - private String columnName; - - private String type; - - private Map<String, String> properties; - - public Field() { - properties = new HashMap<>(2); - } - - public void addProperty(String key, String value) { - properties.put(key, value); - } - - public String getFieldName() { - return fieldName; - } - - public void setFieldName(String fieldName) { - this.fieldName = fieldName; - } - - public String getColumnName() { - return columnName; - } - - public void setColumnName(String columnName) { - this.columnName = columnName; - } - - public String getType() { - return type; - } - - public void setType(String type) { - this.type = type; - } - - public String getProperty(String key) { - return this.properties.get(key); - } -} http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/KeySpace.java ---------------------------------------------------------------------- diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/KeySpace.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/KeySpace.java deleted file mode 100644 index 8a7ca5f..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/KeySpace.java +++ /dev/null @@ -1,84 +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.bean; - -import java.util.HashMap; -import java.util.Map; - -/** - * This class represents the Cassandra Keyspace. - */ -public class KeySpace { - - private String name; - - private PlacementStrategy placementStrategy; - - private boolean durableWritesEnabled; - - private int replicationFactor; - - private Map<String, Integer> dataCenters; - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - public boolean isDurableWritesEnabled() { - return durableWritesEnabled; - } - - public void setDurableWritesEnabled(boolean durableWritesEnabled) { - this.durableWritesEnabled = durableWritesEnabled; - } - - public PlacementStrategy getPlacementStrategy() { - return placementStrategy; - } - - public void setPlacementStrategy(PlacementStrategy placementStrategy) { - this.placementStrategy = placementStrategy; - if (placementStrategy.equals(PlacementStrategy.NetworkTopologyStrategy) && this.dataCenters == null) { - this.dataCenters = new HashMap<>(); - } - } - - public int getReplicationFactor() { - return replicationFactor; - } - - public void setReplicationFactor(int replicationFactor) { - this.replicationFactor = replicationFactor; - } - - public Map<String, Integer> getDataCenters() { - return dataCenters; - } - - public void addDataCenter(String key, Integer value) { - this.dataCenters.put(key, value); - } - - public enum PlacementStrategy { - SimpleStrategy, - NetworkTopologyStrategy, - } -} http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/PartitionKeyField.java ---------------------------------------------------------------------- diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/PartitionKeyField.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/PartitionKeyField.java deleted file mode 100644 index be91d40..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/bean/PartitionKeyField.java +++ /dev/null @@ -1,50 +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.bean; - -import java.util.ArrayList; -import java.util.List; - -/** - * This class represents Cassandra Partition Key. - */ -public class PartitionKeyField extends Field { - - private boolean isComposite; - - private List<Field> fields; - - public boolean isComposite() { - return isComposite; - } - - public void setComposite(boolean composite) { - isComposite = composite; - if (isComposite && fields == null) { - fields = new ArrayList<>(); - } - } - - public void addField(Field field) { - this.fields.add(field); - } - - public List<Field> getFields() { - return fields; - } - -} http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/compiler/GoraCassandraNativeCompiler.java ---------------------------------------------------------------------- diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/compiler/GoraCassandraNativeCompiler.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/compiler/GoraCassandraNativeCompiler.java deleted file mode 100644 index cd0a5d2..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/compiler/GoraCassandraNativeCompiler.java +++ /dev/null @@ -1,326 +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 - * <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.compiler; - -import org.apache.commons.io.FilenameUtils; -import org.apache.gora.cassandra.bean.Field; -import org.apache.gora.cassandra.store.CassandraMapping; -import org.apache.gora.cassandra.store.CassandraMappingBuilder; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.File; -import java.io.FileOutputStream; -import java.io.IOException; -import java.io.OutputStreamWriter; -import java.io.Writer; -import java.nio.charset.Charset; -import java.util.List; -import java.util.Locale; - -/** - * This class generate Java classes for Cassandra Native Serialization. - * <p> - * Generate specific Java classes for defined Gora cassandra mapping. - * Different from the @see org.apache.gora.compiler.GoraCompiler, - * which uses an .avsc or .json schema definition, this compiler - * expects an XML mapping file as input. - */ -public class GoraCassandraNativeCompiler { - - private static final Logger log = LoggerFactory.getLogger(GoraCassandraNativeCompiler.class); - - private Writer out; - private File dest; - - GoraCassandraNativeCompiler(File dest) { - this.dest = dest; - } - - /** - * Start point of the compiler program - * - * @param args the schema file to be compiled and where this should be written - */ - public static void main(String[] args) { - try { - if (args.length < 2) { - log.error("Usage: Compiler <mapping file> <output dir>"); - System.exit(1); - } - compileSchema(new File(args[0]), new File(args[1])); - } catch (Exception e) { - log.error("Something went wrong. Please check the input file.", e.getMessage()); - throw new RuntimeException(e); - } - } - - /** - * Generates Java classes for a mapping. - */ - private static void compileSchema(File src, File dest) throws Exception { - log.info("Compiling {} to {}", src, dest); - GoraCassandraNativeCompiler compiler = new GoraCassandraNativeCompiler(dest); - List<CassandraMapping> mappings = readMappingFile(src); - for (CassandraMapping mapping : mappings) { - compiler.compile(mapping); - } - } - - private static List<CassandraMapping> readMappingFile(File src) throws Exception { - List<CassandraMapping> mappings = new CassandraMappingBuilder().readMappingFile(src); - return mappings; - } - - /** - * Returns the string received with the first letter in uppercase - * - * @param name to be converted - * @return camelCase String - */ - static String cap(String name) { - return name.substring(0, 1).toUpperCase(Locale.getDefault()) - + name.substring(1, name.length()); - } - - /** - * Method in charge of compiling a specific table using a key schema and a set - * of attributes - * - * @param mapping Cassandra Mapping - */ - private void compile(CassandraMapping mapping) { - String fullQualifiedName = mapping.getProperty("name"); - String tableName = mapping.getCoreName(); - String packageName = fullQualifiedName.substring(0, fullQualifiedName.lastIndexOf(".")); - String className = fullQualifiedName.substring(packageName.length() + 1, fullQualifiedName.length()); - String keySpace = mapping.getKeySpace().getName(); - - try { - startFile(className, packageName); - setHeaders(packageName); - line(0, ""); - line(0, "@Table(keyspace = \"" + keySpace + "\", name = \"" + tableName + "\"," + - "readConsistency = \"QUORUM\"," + - "writeConsistency = \"QUORUM\"," + - "caseSensitiveKeyspace = false," + - "caseSensitiveTable = false)"); - line(0, "public class " + className + " implements Persistent {"); - for (Field field : mapping.getFieldList()) { - processFields(field); - processGetterAndSetters(field); - line(2, ""); - } - - setDefaultMethods(2, className); - line(0, "}"); - out.flush(); - out.close(); - } catch (IOException e) { - log.error("Error while compiling table {}", className, e.getMessage()); - throw new RuntimeException(e); - } - } - - /** - * Sets the necessary imports for the generated java class to work - * - * @param namespace Namespace - * @throws IOException - */ - private void setHeaders(String namespace) throws IOException { - if (namespace != null) { - line(0, "package " + namespace + ";\n"); - } - line(0, "import java.util.List;"); - line(0, "import java.util.Set;"); - line(0, "import java.util.Map;"); - line(0, "import java.util.UUID;"); - line(0, "import java.math.BigDecimal;"); - line(0, "import java.math.BigInteger;"); - line(0, "import java.net.InetAddress;"); - line(0, "import java.nio.ByteBuffer;"); - line(0, "import java.util.Date;"); - line(0, ""); - line(0, "import org.apache.avro.Schema.Field;"); - line(0, "import org.apache.gora.persistency.Persistent;"); - line(0, "import org.apache.gora.persistency.Tombstone;"); - line(0, "import com.datastax.driver.mapping.annotations.Column;"); - line(0, "import com.datastax.driver.mapping.annotations.PartitionKey;"); - line(0, "import com.datastax.driver.mapping.annotations.Table;"); - line(0, "import com.datastax.driver.mapping.annotations.Transient;"); - } - - /** - * Starts the java generated class file - * - * @param name Class name - * @throws IOException - */ - private void startFile(String name, String packageName) throws IOException { - String fullDest = FilenameUtils.normalize - (dest.getAbsolutePath() + File.separatorChar + packageName.replace('.', File.separatorChar)); - File dir = new File(fullDest); - if (!dir.exists()) - if (!dir.mkdirs()) - throw new IOException("Unable to create " + dir); - name = cap(name) + ".java"; - out = new OutputStreamWriter(new FileOutputStream(new File(dir, name)), Charset.defaultCharset()); - } - - /** - * Creates default methods inherited from upper classes - * - * @param pIden of spaces used for indentation - * @param className class Name - * @throws IOException - */ - private void setDefaultMethods(int pIden, String className) throws IOException { - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public void clear() { }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public boolean isDirty() { return false; }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public boolean isDirty(int fieldIndex) { return false; }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public boolean isDirty(String field) { return false; }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public void setDirty() { }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public void setDirty(int fieldIndex) { }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public void setDirty(String field) { }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public void clearDirty(int fieldIndex) { }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public void clearDirty(String field) { }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public void clearDirty() { }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public Tombstone getTombstone() { return null; }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public List<Field> getUnmanagedFields() { return null; }"); - line(pIden, "@Transient"); - line(pIden, "@Override"); - line(pIden, "public Persistent newInstance() { return new " + className + "(); }"); - } - - private void processFields(Field field) throws IOException { - String fieldName = field.getFieldName(); - String columnName = field.getColumnName(); - if (Boolean.parseBoolean(field.getProperty("primarykey"))) { - line(2, "@PartitionKey"); - } - line(2, "@Column(name = \"" + columnName + "\")"); - line(2, "private " + getDataType(field.getType(), false) + " " + fieldName + ";"); - } - - private void processGetterAndSetters(Field field) throws IOException { - String dataType = getDataType(field.getType(), false); - line(2, "public " + dataType + " get" + cap(field.getFieldName()) + "() {"); - line(2, "return " + field.getFieldName() + ";"); - line(2, "}"); - line(2, "public void set" + cap(field.getFieldName()) + "(" + dataType + " field) {"); - line(2, field.getFieldName() + " = field;"); - line(2, "}"); - } - - private String getDataType(String dbType, boolean isInner) { - if (dbType.equalsIgnoreCase("uuid")) { - return "UUID"; - } else if (dbType.equalsIgnoreCase("text") || dbType.equalsIgnoreCase("ascii") || dbType.equalsIgnoreCase("varchar")) { - return "String"; - } else if (dbType.equalsIgnoreCase("timestamp")) { - return "Date"; - } else if (dbType.startsWith("list")) { - String innerType = dbType.substring(dbType.indexOf("<") + 1, dbType.indexOf(">")); - return "List<" + getDataType(innerType, true) + ">"; - } else if (dbType.startsWith("set")) { - String innerType = dbType.substring(dbType.indexOf("<") + 1, dbType.indexOf(">")); - return "Set<" + getDataType(innerType, true) + ">"; - } else if (dbType.startsWith("map")) { - String innerTypes = dbType.substring(dbType.indexOf("<") + 1, dbType.indexOf(">")); - String[] types = innerTypes.split(","); - return "Map<" + getDataType(types[0], true) + "," + getDataType(types[1], true) + ">"; - } else if (dbType.equalsIgnoreCase("blob")) { - return "ByteBuffer"; - } else if (dbType.equalsIgnoreCase("int")) { - if (isInner) { - return "Integer"; - } else { - return "int"; - } - } else if (dbType.equalsIgnoreCase("float")) { - if (isInner) { - return "Float"; - } else { - return "float"; - } - } else if (dbType.equalsIgnoreCase("double")) { - if (isInner) { - return "Double"; - } else { - return "double"; - } - } else if (dbType.equalsIgnoreCase("decimal")) { - return "BigDecimal"; - } else if (dbType.equalsIgnoreCase("bigint") || dbType.equalsIgnoreCase("counter")) { - return "Long"; - } else if (dbType.equalsIgnoreCase("boolean")) { - if (isInner) { - return "Boolean"; - } else { - return "boolean"; - } - } else if (dbType.equalsIgnoreCase("varint")) { - return "BigInteger"; - } else if (dbType.equalsIgnoreCase("inet")) { - return "InetAddress"; - } else if (dbType.contains("frozen")) { - throw new RuntimeException("Compiler Doesn't support user define types"); - } - throw new RuntimeException("Invalid Cassandra DataType"); - } - - /** - * Writes a line within the output stream - * - * @param indent Number of spaces used for indentation - * @param text Text to be written - * @throws IOException - */ - private void line(int indent, String text) throws IOException { - for (int i = 0; i < indent; i++) { - out.append(" "); - } - out.append(text); - out.append("\n"); - } -} http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/package-info.java ---------------------------------------------------------------------- diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/package-info.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/package-info.java deleted file mode 100644 index 3ad9186..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/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 - * <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. - */ - -/** - * 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/89683c74/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java ---------------------------------------------------------------------- diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java deleted file mode 100644 index e24822d..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.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 - * <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.query; - -import org.apache.gora.filter.Filter; -import org.apache.gora.persistency.Persistent; -import org.apache.gora.query.Query; -import org.apache.gora.query.ws.impl.QueryWSBase; -import org.apache.gora.store.DataStore; - -import java.util.HashMap; -import java.util.Map; - -/** - * Cassandra specific implementation of the {@link Query} interface. - */ -public class CassandraQuery<K, T extends Persistent> extends QueryWSBase<K, T> { - - private Filter<K, T> filter; - - private boolean localFilterEnabled = true; - - private Map<String, Object> updateFields = new HashMap<>(); - - public CassandraQuery(DataStore<K, T> dataStore) { - super(dataStore); - } - - /** - * {@inheritDoc} - */ - @Override - public Filter<K, T> getFilter() { - return filter; - } - - /** - * {@inheritDoc} - */ - @Override - public void setFilter(Filter<K, T> filter) { - this.filter = filter; - } - - /** - * {@inheritDoc} - */ - @Override - public boolean isLocalFilterEnabled() { - return localFilterEnabled; - } - - /** - * {@inheritDoc} - */ - @Override - public void setLocalFilterEnabled(boolean enable) { - localFilterEnabled = enable; - } - - /** - * This method adds Update field with the relevant Value - * - * @param field field Name - * @param newValue New Value of the field - */ - public void addUpdateField(String field, Object newValue) { - updateFields.put(field, newValue); - } - - /** - * This method returns the updated field value of the particular field. - * - * @param key Field Name - * @return Object value - */ - public Object getUpdateFieldValue(String key) { - return updateFields.get(key); - } - - /** - * {@inheritDoc} - */ - @Override - public String[] getFields() { - if (updateFields.size() == 0) { - return super.getFields(); - } else { - String[] updateFieldsArray = new String[updateFields.size()]; - return updateFields.keySet().toArray(updateFieldsArray); - } - } -} http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/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 deleted file mode 100644 index 4e44d0d..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/CassandraResultSet.java +++ /dev/null @@ -1,121 +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 - * <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.query; - -import org.apache.gora.persistency.Persistent; -import org.apache.gora.query.Query; -import org.apache.gora.query.impl.ResultBase; -import org.apache.gora.store.DataStore; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; - -/** - * CassandraResult specific implementation of the {@link org.apache.gora.query.Result} - * interface. - */ -public class CassandraResultSet<K, T extends Persistent> extends ResultBase<K, T> { - - private List<T> persistentObject = new ArrayList<T>(); - - private List<K> persistentKey = new ArrayList<K>(); - - private int size = 0; - - private int position = 0; - - /** - * Constructor of the Cassandra Result - * @param dataStore Cassandra Data Store - * @param query Cassandra Query - */ - public CassandraResultSet(DataStore<K, T> dataStore, Query<K, T> query) { - super(dataStore, query); - } - - /** - * {@inheritDoc} - * - * @return - * @throws IOException - */ - @Override - protected boolean nextInner() throws IOException { - if (offset < size) { - persistent = persistentObject.get(position); - key = persistentKey.get(position); - position++; - return true; - } - return false; - } - - /** - * {@inheritDoc} - * - * @return - * @throws IOException - * @throws InterruptedException - */ - @Override - public float getProgress() throws IOException, InterruptedException { - return ((float) position) / size; - } - - /** - * {@inheritDoc} - * - * @return - */ - @Override - public T get() { - return super.get(); - } - - /** - * {@inheritDoc} - * - * @return - */ - @Override - public K getKey() { - return super.getKey(); - } - - /** - * 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); - this.persistentObject.add(token); - this.size++; - } - - @Override - /** - * Returns whether the limit for the query is reached. - * @return true if result limit is reached - */ - protected boolean isLimitReached() { - return (limit > 0 && offset >= limit) || (offset >= size); - } -} http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/package-info.java ---------------------------------------------------------------------- diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/package-info.java b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/package-info.java deleted file mode 100644 index 275c8d9..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/query/package-info.java +++ /dev/null @@ -1,22 +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 - * <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. - */ - -/** - * 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/89683c74/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 deleted file mode 100644 index 9c33bf6..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroCassandraUtils.java +++ /dev/null @@ -1,354 +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 - * <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.serializers; - -import org.apache.avro.Schema; -import org.apache.avro.util.Utf8; -import org.apache.gora.cassandra.bean.CassandraKey; -import org.apache.gora.cassandra.bean.Field; -import org.apache.gora.cassandra.store.CassandraMapping; -import org.apache.gora.hbase.util.HBaseByteInterface; -import org.apache.gora.persistency.Persistent; -import org.apache.gora.persistency.impl.DirtyListWrapper; -import org.apache.gora.persistency.impl.DirtyMapWrapper; -import org.apache.gora.persistency.impl.PersistentBase; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.math.BigDecimal; -import java.math.BigInteger; -import java.net.InetAddress; -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Date; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.UUID; - -/** - * This class is Utils class for Avro serialization. - */ -class AvroCassandraUtils { - - /** - * Default schema index with value "0" used when AVRO Union data types are stored. - */ - private static final int DEFAULT_UNION_SCHEMA = 0; - - private static final Logger LOG = LoggerFactory.getLogger(AvroCassandraUtils.class); - - static void processKeys(CassandraMapping cassandraMapping, Object key, List<String> keys, List<Object> values) { - CassandraKey cassandraKey = cassandraMapping.getCassandraKey(); - if (cassandraKey != null) { - if (key instanceof PersistentBase) { - PersistentBase keyBase = (PersistentBase) key; - for (Schema.Field field : keyBase.getSchema().getFields()) { - Field mappedField = cassandraKey.getFieldFromFieldName(field.name()); - if (mappedField != null) { - keys.add(field.name()); - Object value = keyBase.get(field.pos()); - value = getFieldValueFromAvroBean(field.schema(), field.schema().getType(), value, mappedField); - values.add(value); - } else { - LOG.debug("Ignoring field {}, Since field couldn't find in the {} mapping", new Object[]{field.name(), cassandraMapping.getPersistentClass()}); - } - } - } else { - LOG.error("Key bean isn't extended by {} .", new Object[]{cassandraMapping.getKeyClass(), PersistentBase.class}); - } - } else { - keys.add(cassandraMapping.getInlinedDefinedPartitionKey().getFieldName()); - values.add(key); - } - } - - /** - * For every field within an object, we pass in a field schema, Type and value. - * This enables us to process fields (based on their characteristics) - * preparing them for persistence. - * - * @param fieldSchema the associated field schema - * @param type the field type - * @param fieldValue the field value. - * @return field value - */ - static Object getFieldValueFromAvroBean(Schema fieldSchema, Schema.Type type, Object fieldValue, Field field) { - switch (type) { - // Record can be persist with two ways, udt and bytes - case RECORD: - if (field.getType().contains("blob")) { - try { - byte[] serializedBytes = HBaseByteInterface.toBytes(fieldValue, fieldSchema); - fieldValue = ByteBuffer.wrap(serializedBytes); - } catch (IOException e) { - LOG.error("Error occurred when serializing {} field. {}", new Object[]{field.getFieldName(), e.getMessage()}); - } - } else { - 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(); - Map<String, Object> map = new HashMap<>(); - for (Map.Entry<CharSequence, ?> e : ((Map<CharSequence, ?>) fieldValue).entrySet()) { - String mapKey = e.getKey().toString(); - Object mapValue = e.getValue(); - mapValue = getFieldValueFromAvroBean(valueSchema, valuetype, mapValue, field); - map.put(mapKey, mapValue); - } - fieldValue = map; - break; - case ARRAY: - valueSchema = fieldSchema.getElementType(); - valuetype = valueSchema.getType(); - ArrayList<Object> list = new ArrayList<>(); - for (Object item : (Collection<?>) fieldValue) { - Object value = getFieldValueFromAvroBean(valueSchema, valuetype, item, field); - list.add(value); - } - fieldValue = list; - break; - case UNION: - // storing the union selected schema, the actual value will - // be stored as soon as we get break out. - if (fieldValue != null) { - int schemaPos = getUnionSchema(fieldValue, fieldSchema); - Schema unionSchema = fieldSchema.getTypes().get(schemaPos); - Schema.Type unionType = unionSchema.getType(); - fieldValue = getFieldValueFromAvroBean(unionSchema, unionType, fieldValue, field); - } - break; - case STRING: - fieldValue = fieldValue.toString(); - break; - default: - break; - } - return fieldValue; - } - - /** - * Given an object and the object schema this function obtains, - * from within the UNION schema, the position of the type used. - * If no data type can be inferred then we return a default value - * of position 0. - * - * @param pValue Object - * @param pUnionSchema avro Schema - * @return the unionSchemaPosition. - */ - private static int getUnionSchema(Object pValue, Schema pUnionSchema) { - int unionSchemaPos = 0; -// String valueType = pValue.getClass().getSimpleName(); - for (Schema currentSchema : pUnionSchema.getTypes()) { - Schema.Type schemaType = currentSchema.getType(); - if (pValue instanceof CharSequence && schemaType.equals(Schema.Type.STRING)) - return unionSchemaPos; - else if (pValue instanceof ByteBuffer && schemaType.equals(Schema.Type.BYTES)) - return unionSchemaPos; - else if (pValue instanceof Integer && schemaType.equals(Schema.Type.INT)) - return unionSchemaPos; - else if (pValue instanceof Long && schemaType.equals(Schema.Type.LONG)) - return unionSchemaPos; - else if (pValue instanceof Double && schemaType.equals(Schema.Type.DOUBLE)) - return unionSchemaPos; - else if (pValue instanceof Float && schemaType.equals(Schema.Type.FLOAT)) - return unionSchemaPos; - else if (pValue instanceof Boolean && schemaType.equals(Schema.Type.BOOLEAN)) - return unionSchemaPos; - else if (pValue instanceof Map && schemaType.equals(Schema.Type.MAP)) - return unionSchemaPos; - else if (pValue instanceof List && schemaType.equals(Schema.Type.ARRAY)) - return unionSchemaPos; - else if (pValue instanceof Persistent && schemaType.equals(Schema.Type.RECORD)) - return unionSchemaPos; - else if (pValue != null && ByteBuffer.class.isAssignableFrom(pValue.getClass()) && schemaType.equals(Schema.Type.STRING)) - return unionSchemaPos; - else if (pValue != null && ByteBuffer.class.isAssignableFrom(pValue.getClass()) && schemaType.equals(Schema.Type.INT)) - return unionSchemaPos; - else if (pValue != null && ByteBuffer.class.isAssignableFrom(pValue.getClass()) && schemaType.equals(Schema.Type.LONG)) - return unionSchemaPos; - else if (pValue != null && ByteBuffer.class.isAssignableFrom(pValue.getClass()) && schemaType.equals(Schema.Type.DOUBLE)) - return unionSchemaPos; - else if (pValue != null && ByteBuffer.class.isAssignableFrom(pValue.getClass()) && schemaType.equals(Schema.Type.FLOAT)) - return unionSchemaPos; - else if (pValue != null && ByteBuffer.class.isAssignableFrom(pValue.getClass()) && schemaType.equals(Schema.Type.RECORD)) - return unionSchemaPos; - unionSchemaPos++; - } - // if we weren't able to determine which data type it is, then we return the default - return DEFAULT_UNION_SCHEMA; - } - - static Object getAvroFieldValue(Object value, Schema schema) { - Object result; - switch (schema.getType()) { - - case MAP: - Map<String, Object> rawMap = (Map<String, Object>) value; - Map<Utf8, Object> utf8ObjectHashMap = new HashMap<>(); - if (rawMap == null) { - result = new DirtyMapWrapper(utf8ObjectHashMap); - break; - } - for (Map.Entry<?, ?> e : rawMap.entrySet()) { - Schema innerSchema = schema.getValueType(); - Object obj = getAvroFieldValue(e.getValue(), innerSchema); - if (e.getKey().getClass().getSimpleName().equalsIgnoreCase("Utf8")) { - utf8ObjectHashMap.put((Utf8) e.getKey(), obj); - } else { - utf8ObjectHashMap.put(new Utf8((String) e.getKey()), obj); - } - } - result = new DirtyMapWrapper<>(utf8ObjectHashMap); - break; - - case ARRAY: - List<Object> rawList = (List<Object>) value; - List<Object> objectArrayList = new ArrayList<>(); - if (rawList == null) { - return new DirtyListWrapper(objectArrayList); - } - for (Object item : rawList) { - Object obj = getAvroFieldValue(item, schema.getElementType()); - objectArrayList.add(obj); - } - result = new DirtyListWrapper<>(objectArrayList); - break; - - case RECORD: - if (value != null && ByteBuffer.class.isAssignableFrom(value.getClass())) { - ByteBuffer buffer = (ByteBuffer) value; - byte[] arr = new byte[buffer.remaining()]; - buffer.get(arr); - try { - result = (PersistentBase) HBaseByteInterface.fromBytes(schema, arr); - } catch (IOException e) { - LOG.error("Error occurred while deserialize the Record. :" + e.getMessage()); - result = null; - } - } else { - result = (PersistentBase) value; - } - break; - - case UNION: - int index = getUnionSchema(value, schema); - Schema resolvedSchema = schema.getTypes().get(index); - result = getAvroFieldValue(value, resolvedSchema); - break; - - case ENUM: - result = org.apache.gora.util.AvroUtils.getEnumValue(schema, (String) value); - break; - - case BYTES: - if (ByteBuffer.class.isAssignableFrom(value.getClass())) { - result = value; - } else { - result = ByteBuffer.wrap((byte[]) value); - } - break; - - case STRING: - if (value instanceof org.apache.avro.util.Utf8) { - result = value; - } else if (ByteBuffer.class.isAssignableFrom(value.getClass())) { - result = new Utf8(((ByteBuffer) value).array()); - } else { - result = new Utf8((String) value); - } - break; - - case INT: - if (ByteBuffer.class.isAssignableFrom(value.getClass())) { - result = ((ByteBuffer) value).getInt(); - } else { - result = value; - } - break; - - case FLOAT: - if (ByteBuffer.class.isAssignableFrom(value.getClass())) { - result = ((ByteBuffer) value).getFloat(); - } else { - result = value; - } - break; - - case DOUBLE: - if (ByteBuffer.class.isAssignableFrom(value.getClass())) { - result = ((ByteBuffer) value).getDouble(); - } else { - result = value; - } - break; - - case LONG: - if (ByteBuffer.class.isAssignableFrom(value.getClass())) { - result = ((ByteBuffer) value).getLong(); - } else { - result = value; - } - break; - - default: - result = value; - } - return result; - } - - static Class getRelevantClassForCassandraDataType(String dataType) { - switch (dataType) { - 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"); - } - } - -} http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/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 deleted file mode 100644 index 204ae52..0000000 --- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializer.java +++ /dev/null @@ -1,446 +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 com.datastax.driver.core.AbstractGettableData; -import com.datastax.driver.core.ColumnDefinitions; -import com.datastax.driver.core.ConsistencyLevel; -import com.datastax.driver.core.DataType; -import com.datastax.driver.core.ResultSet; -import com.datastax.driver.core.Row; -import com.datastax.driver.core.SimpleStatement; -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; -import org.apache.gora.cassandra.query.CassandraResultSet; -import org.apache.gora.cassandra.store.CassandraClient; -import org.apache.gora.cassandra.store.CassandraMapping; -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; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.Iterator; -import java.util.List; -import java.util.Map; - -/** - * This class contains the operations relates to Avro Serialization. - */ -class AvroSerializer<K, T extends PersistentBase> extends CassandraSerializer { - - - private static final Logger LOG = LoggerFactory.getLogger(AvroSerializer.class); - - private DataStore<K, T> cassandraDataStore; - - private Schema persistentSchema; - - AvroSerializer(CassandraClient cassandraClient, DataStore<K, T> dataStore, CassandraMapping mapping) { - super(cassandraClient, dataStore.getKeyClass(), dataStore.getPersistentClass(), mapping); - if (PersistentBase.class.isAssignableFrom(dataStore.getPersistentClass())) { - persistentSchema = ((PersistentBase) dataStore.getBeanFactory().getCachedPersistent()).getSchema(); - } else { - persistentSchema = null; - } - this.cassandraDataStore = dataStore; - try { - analyzePersistent(); - } catch (Exception e) { - throw new RuntimeException("Error occurred while analyzing the persistent class, :" + e.getMessage()); - } - } - - /** - * {@inheritDoc} - * - * @throws Exception - */ - protected 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 (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); - } - } - } - } - - /** - * {@inheritDoc} - * - * @param query - * @return - */ - @Override - public boolean updateByQuery(Query query) { - List<Object> objectArrayList = new ArrayList<>(); - String cqlQuery = CassandraQueryFactory.getUpdateByQueryForAvro(mapping, query, objectArrayList, persistentSchema); - ResultSet results; - SimpleStatement statement; - if (objectArrayList.size() == 0) { - statement = new SimpleStatement(cqlQuery); - } else { - statement = new SimpleStatement(cqlQuery, objectArrayList.toArray()); - } - if (writeConsistencyLevel != null) { - statement.setConsistencyLevel(ConsistencyLevel.valueOf(writeConsistencyLevel)); - } - results = client.getSession().execute(statement); - return results.wasApplied(); - } - - /** - * {@inheritDoc} - * - * @param key - * @param fields - * @return - */ - @Override - public Persistent get(Object key, String[] fields) { - if (fields == null) { - fields = getFields(); - } - ArrayList<String> cassandraKeys = new ArrayList<>(); - ArrayList<Object> cassandraValues = new ArrayList<>(); - AvroCassandraUtils.processKeys(mapping, key, cassandraKeys, cassandraValues); - String cqlQuery = CassandraQueryFactory.getSelectObjectWithFieldsQuery(mapping, fields, cassandraKeys); - SimpleStatement statement = new SimpleStatement(cqlQuery, cassandraValues.toArray()); - if (readConsistencyLevel != null) { - statement.setConsistencyLevel(ConsistencyLevel.valueOf(readConsistencyLevel)); - } - ResultSet resultSet = this.client.getSession().execute(statement); - Iterator<Row> iterator = resultSet.iterator(); - ColumnDefinitions definitions = resultSet.getColumnDefinitions(); - T obj = null; - if (iterator.hasNext()) { - obj = cassandraDataStore.newPersistent(); - AbstractGettableData row = (AbstractGettableData) iterator.next(); - populateValuesToPersistent(row, definitions, obj, fields); - } - return obj; - } - - /** - * {@inheritDoc} - * - * @param key - * @param persistent - */ - @Override - public void put(Object key, Persistent persistent) { - if (persistent instanceof PersistentBase) { - if (persistent.isDirty()) { - PersistentBase persistentBase = (PersistentBase) persistent; - ArrayList<String> fields = new ArrayList<>(); - ArrayList<Object> values = new ArrayList<>(); - AvroCassandraUtils.processKeys(mapping, key, fields, values); - for (Schema.Field f : persistentBase.getSchema().getFields()) { - String fieldName = f.name(); - Field field = mapping.getFieldFromFieldName(fieldName); - if (field == null) { - LOG.debug("Ignoring {} adding field, {} field can't find in {} mapping", new Object[]{fieldName, fieldName, persistentClass}); - continue; - } - if (persistent.isDirty(f.pos()) || mapping.getInlinedDefinedPartitionKey().equals(mapping.getFieldFromFieldName(fieldName))) { - Object value = persistentBase.get(f.pos()); - 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); - } - } - String cqlQuery = CassandraQueryFactory.getInsertDataQuery(mapping, fields); - SimpleStatement statement = new SimpleStatement(cqlQuery, values.toArray()); - if (writeConsistencyLevel != null) { - statement.setConsistencyLevel(ConsistencyLevel.valueOf(writeConsistencyLevel)); - } - client.getSession().execute(statement); - } else { - LOG.info("Ignored putting persistent bean {} in the store as it is neither " - + "new, neither dirty.", new Object[]{persistent}); - } - } else { - LOG.error("{} Persistent bean isn't extended by {} .", new Object[]{this.persistentClass, PersistentBase.class}); - } - } - - /** - * {@inheritDoc} - * - * @param key - * @return - */ - @Override - public Persistent get(Object key) { - ArrayList<String> cassandraKeys = new ArrayList<>(); - ArrayList<Object> cassandraValues = new ArrayList<>(); - AvroCassandraUtils.processKeys(mapping, key, cassandraKeys, cassandraValues); - String cqlQuery = CassandraQueryFactory.getSelectObjectQuery(mapping, cassandraKeys); - SimpleStatement statement = new SimpleStatement(cqlQuery, cassandraValues.toArray()); - if (readConsistencyLevel != null) { - statement.setConsistencyLevel(ConsistencyLevel.valueOf(readConsistencyLevel)); - } - ResultSet resultSet = client.getSession().execute(statement); - Iterator<Row> iterator = resultSet.iterator(); - ColumnDefinitions definitions = resultSet.getColumnDefinitions(); - T obj = null; - if (iterator.hasNext()) { - obj = cassandraDataStore.newPersistent(); - AbstractGettableData row = (AbstractGettableData) iterator.next(); - populateValuesToPersistent(row, definitions, obj, mapping.getFieldNames()); - } - return obj; - } - - /** - * This method wraps result set data in to DataEntry and creates a list of DataEntry. - **/ - private void populateValuesToPersistent(AbstractGettableData row, ColumnDefinitions columnDefinitions, PersistentBase base, String[] fields) { - Object paramValue; - for (String fieldName : fields) { - Schema.Field avroField = base.getSchema().getField(fieldName); - Field field = mapping.getFieldFromFieldName(fieldName); - //to ignore unspecified fields in the mapping - if (field == null || avroField == null) { - continue; - } - Schema fieldSchema = avroField.schema(); - String columnName = field.getColumnName(); - paramValue = getValue(row, columnDefinitions.getType(columnName), columnName, fieldSchema); - Object value = AvroCassandraUtils.getAvroFieldValue(paramValue, fieldSchema); - base.put(avroField.pos(), value); - } - } - - private Object getValue(AbstractGettableData row, DataType columnType, String columnName, Schema schema) { - Object paramValue; - String dataType; - switch (columnType.getName()) { - case ASCII: - paramValue = row.getString(columnName); - break; - case BIGINT: - paramValue = row.isNull(columnName) ? null : row.getLong(columnName); - break; - case BLOB: - paramValue = row.isNull(columnName) ? null : row.getBytes(columnName); - break; - case BOOLEAN: - paramValue = row.isNull(columnName) ? null : row.getBool(columnName); - break; - case COUNTER: - paramValue = row.isNull(columnName) ? null : row.getLong(columnName); - break; - case DECIMAL: - paramValue = row.isNull(columnName) ? null : row.getDecimal(columnName); - break; - case DOUBLE: - paramValue = row.isNull(columnName) ? null : row.getDouble(columnName); - break; - case FLOAT: - paramValue = row.isNull(columnName) ? null : row.getFloat(columnName); - break; - case INET: - paramValue = row.isNull(columnName) ? null : row.getInet(columnName).toString(); - break; - case INT: - paramValue = row.isNull(columnName) ? null : row.getInt(columnName); - break; - case TEXT: - paramValue = row.getString(columnName); - break; - case TIMESTAMP: - paramValue = row.isNull(columnName) ? null : row.getDate(columnName); - break; - case UUID: - paramValue = row.isNull(columnName) ? null : row.getUUID(columnName); - break; - case VARCHAR: - paramValue = row.getString(columnName); - break; - case VARINT: - paramValue = row.isNull(columnName) ? null : row.getVarint(columnName); - break; - case TIMEUUID: - paramValue = row.isNull(columnName) ? null : row.getUUID(columnName); - break; - case LIST: - dataType = columnType.getTypeArguments().get(0).toString(); - paramValue = row.isNull(columnName) ? null : row.getList(columnName, AvroCassandraUtils.getRelevantClassForCassandraDataType(dataType)); - break; - case SET: - dataType = columnType.getTypeArguments().get(0).toString(); - paramValue = row.isNull(columnName) ? null : row.getList(columnName, AvroCassandraUtils.getRelevantClassForCassandraDataType(dataType)); - break; - case MAP: - dataType = columnType.getTypeArguments().get(1).toString(); - // Avro supports only String for keys - paramValue = row.isNull(columnName) ? null : row.getMap(columnName, String.class, AvroCassandraUtils.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(); - break; - case CUSTOM: - paramValue = row.isNull(columnName) ? null : row.getBytes(columnName); - break; - default: - paramValue = row.getString(columnName); - break; - } - return paramValue; - } - - /** - * {@inheritDoc} - * - * @param key - * @return - */ - @Override - public boolean delete(Object key) { - ArrayList<String> cassandraKeys = new ArrayList<>(); - ArrayList<Object> cassandraValues = new ArrayList<>(); - AvroCassandraUtils.processKeys(mapping, key, cassandraKeys, cassandraValues); - String cqlQuery = CassandraQueryFactory.getDeleteDataQuery(mapping, cassandraKeys); - SimpleStatement statement = new SimpleStatement(cqlQuery, cassandraValues.toArray()); - if (writeConsistencyLevel != null) { - statement.setConsistencyLevel(ConsistencyLevel.valueOf(writeConsistencyLevel)); - } - ResultSet resultSet = client.getSession().execute(statement); - return resultSet.wasApplied(); - } - - /** - * {@inheritDoc} - * - * @param dataStore - * @param query - * @return - */ - @Override - public Result execute(DataStore dataStore, Query query) { - List<Object> objectArrayList = new ArrayList<>(); - String[] fields = query.getFields(); - if (fields != null) { - fields = (String[]) ArrayUtils.addAll(fields, mapping.getAllKeys()); - } else { - fields = mapping.getAllFieldsIncludingKeys(); - } - CassandraResultSet<K, T> cassandraResult = new CassandraResultSet<>(dataStore, query); - String cqlQuery = CassandraQueryFactory.getExecuteQuery(mapping, query, objectArrayList, fields); - ResultSet results; - SimpleStatement statement; - if (objectArrayList.size() == 0) { - statement = new SimpleStatement(cqlQuery); - } else { - statement = new SimpleStatement(cqlQuery, objectArrayList.toArray()); - } - if (readConsistencyLevel != null) { - statement.setConsistencyLevel(ConsistencyLevel.valueOf(readConsistencyLevel)); - } - results = client.getSession().execute(statement); - Iterator<Row> iterator = results.iterator(); - ColumnDefinitions definitions = results.getColumnDefinitions(); - T obj; - K keyObject; - CassandraKey cassandraKey = mapping.getCassandraKey(); - while (iterator.hasNext()) { - AbstractGettableData row = (AbstractGettableData) iterator.next(); - obj = cassandraDataStore.newPersistent(); - keyObject = cassandraDataStore.newKey(); - populateValuesToPersistent(row, definitions, obj, fields); - if (cassandraKey != null) { - populateValuesToPersistent(row, definitions, (PersistentBase) keyObject, cassandraKey.getFieldNames()); - } else { - Field key = mapping.getInlinedDefinedPartitionKey(); - keyObject = (K) getValue(row, definitions.getType(key.getColumnName()), key.getColumnName(), null); - } - cassandraResult.addResultElement(keyObject, obj); - } - return cassandraResult; - } - -}
