diff --git a/core/src/integration-test/java/com/scalar/db/storage/cassandra/CassandraCrossPartitionScanIntegrationTest.java b/core/src/integration-test/java/com/scalar/db/storage/cassandra/CassandraCrossPartitionScanIntegrationTest.java new file mode 100644 index 0000000000..d2ace88450 --- /dev/null +++ b/core/src/integration-test/java/com/scalar/db/storage/cassandra/CassandraCrossPartitionScanIntegrationTest.java @@ -0,0 +1,27 @@ +package com.scalar.db.storage.cassandra; + +import com.scalar.db.api.DistributedStorageCrossPartitionScanIntegrationTestBase; +import java.util.Collections; +import java.util.Map; +import java.util.Properties; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +public class CassandraCrossPartitionScanIntegrationTest + extends DistributedStorageCrossPartitionScanIntegrationTestBase { + + @Override + protected Properties getProperties(String testName) { + return CassandraEnv.getProperties(testName); + } + + @Override + protected Map getCreationOptions() { + return Collections.singletonMap(CassandraAdmin.REPLICATION_FACTOR, "1"); + } + + @Test + @Override + @Disabled("Cross partition scan with ordering is not supported in Cassandra") + public void scan_WithOrderingForNonPrimaryColumns_ShouldReturnProperResult() {} +} diff --git a/core/src/integration-test/java/com/scalar/db/storage/cassandra/CassandraEnv.java b/core/src/integration-test/java/com/scalar/db/storage/cassandra/CassandraEnv.java index b542291604..4af38f013e 100644 --- a/core/src/integration-test/java/com/scalar/db/storage/cassandra/CassandraEnv.java +++ b/core/src/integration-test/java/com/scalar/db/storage/cassandra/CassandraEnv.java @@ -25,7 +25,7 @@ public static Properties getProperties(String testName) { properties.setProperty(DatabaseConfig.USERNAME, username); properties.setProperty(DatabaseConfig.PASSWORD, password); properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN, "true"); - properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_FILTERING, "false"); + properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_FILTERING, "true"); properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_ORDERING, "false"); // Add testName as a metadata schema suffix diff --git a/core/src/integration-test/java/com/scalar/db/storage/cassandra/ConsensusCommitCrossPartitionScanIntegrationTestWithCassandra.java b/core/src/integration-test/java/com/scalar/db/storage/cassandra/ConsensusCommitCrossPartitionScanIntegrationTestWithCassandra.java new file mode 100644 index 0000000000..5d4d6289d0 --- /dev/null +++ b/core/src/integration-test/java/com/scalar/db/storage/cassandra/ConsensusCommitCrossPartitionScanIntegrationTestWithCassandra.java @@ -0,0 +1,30 @@ +package com.scalar.db.storage.cassandra; + +import com.scalar.db.transaction.consensuscommit.ConsensusCommitConfig; +import com.scalar.db.transaction.consensuscommit.ConsensusCommitCrossPartitionScanIntegrationTestBase; +import java.util.Collections; +import java.util.Map; +import java.util.Properties; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +public class ConsensusCommitCrossPartitionScanIntegrationTestWithCassandra + extends ConsensusCommitCrossPartitionScanIntegrationTestBase { + + @Override + protected Properties getProps(String testName) { + Properties properties = CassandraEnv.getProperties(testName); + properties.setProperty(ConsensusCommitConfig.ISOLATION_LEVEL, "SERIALIZABLE"); + return properties; + } + + @Override + protected Map getCreationOptions() { + return Collections.singletonMap(CassandraAdmin.REPLICATION_FACTOR, "1"); + } + + @Test + @Override + @Disabled("Cross partition scan with ordering is not supported in Cassandra") + public void scan_CrossPartitionScanWithOrderingGivenForCommittedRecord_ShouldReturnRecords() {} +} diff --git a/core/src/integration-test/java/com/scalar/db/storage/cassandra/TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithCassandra.java b/core/src/integration-test/java/com/scalar/db/storage/cassandra/TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithCassandra.java new file mode 100644 index 0000000000..4109883089 --- /dev/null +++ b/core/src/integration-test/java/com/scalar/db/storage/cassandra/TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithCassandra.java @@ -0,0 +1,30 @@ +package com.scalar.db.storage.cassandra; + +import com.scalar.db.transaction.consensuscommit.ConsensusCommitConfig; +import com.scalar.db.transaction.consensuscommit.TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestBase; +import java.util.Collections; +import java.util.Map; +import java.util.Properties; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +public class TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithCassandra + extends TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestBase { + + @Override + protected Properties getProps1(String testName) { + Properties properties = CassandraEnv.getProperties(testName); + properties.setProperty(ConsensusCommitConfig.ISOLATION_LEVEL, "SERIALIZABLE"); + return properties; + } + + @Override + protected Map getCreationOptions() { + return Collections.singletonMap(CassandraAdmin.REPLICATION_FACTOR, "1"); + } + + @Test + @Override + @Disabled("Cross partition scan with ordering is not supported in Cassandra") + public void scan_ScanWithOrderingGivenForCommittedRecord_ShouldReturnRecords() {} +} diff --git a/core/src/integration-test/java/com/scalar/db/storage/cosmos/ConsensusCommitCrossPartitionScanIntegrationTestWithCosmos.java b/core/src/integration-test/java/com/scalar/db/storage/cosmos/ConsensusCommitCrossPartitionScanIntegrationTestWithCosmos.java new file mode 100644 index 0000000000..73f6ba7578 --- /dev/null +++ b/core/src/integration-test/java/com/scalar/db/storage/cosmos/ConsensusCommitCrossPartitionScanIntegrationTestWithCosmos.java @@ -0,0 +1,29 @@ +package com.scalar.db.storage.cosmos; + +import com.scalar.db.transaction.consensuscommit.ConsensusCommitConfig; +import com.scalar.db.transaction.consensuscommit.ConsensusCommitCrossPartitionScanIntegrationTestBase; +import java.util.Map; +import java.util.Properties; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +public class ConsensusCommitCrossPartitionScanIntegrationTestWithCosmos + extends ConsensusCommitCrossPartitionScanIntegrationTestBase { + + @Override + protected Properties getProps(String testName) { + Properties properties = CosmosEnv.getProperties(testName); + properties.setProperty(ConsensusCommitConfig.ISOLATION_LEVEL, "SERIALIZABLE"); + return properties; + } + + @Override + protected Map getCreationOptions() { + return CosmosEnv.getCreationOptions(); + } + + @Test + @Override + @Disabled("Cross partition scan with ordering is not supported in Cosmos DB") + public void scan_CrossPartitionScanWithOrderingGivenForCommittedRecord_ShouldReturnRecords() {} +} diff --git a/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosCrossPartitionScanIntegrationTest.java b/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosCrossPartitionScanIntegrationTest.java new file mode 100644 index 0000000000..49a4a8b55d --- /dev/null +++ b/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosCrossPartitionScanIntegrationTest.java @@ -0,0 +1,26 @@ +package com.scalar.db.storage.cosmos; + +import com.scalar.db.api.DistributedStorageCrossPartitionScanIntegrationTestBase; +import java.util.Map; +import java.util.Properties; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +public class CosmosCrossPartitionScanIntegrationTest + extends DistributedStorageCrossPartitionScanIntegrationTestBase { + + @Override + protected Properties getProperties(String testName) { + return CosmosEnv.getProperties(testName); + } + + @Override + protected Map getCreationOptions() { + return CosmosEnv.getCreationOptions(); + } + + @Test + @Override + @Disabled("Cross partition scan with ordering is not supported in Cosmos DB") + public void scan_WithOrderingForNonPrimaryColumns_ShouldReturnProperResult() {} +} diff --git a/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosEnv.java b/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosEnv.java index cd91f97b98..ebf5238541 100644 --- a/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosEnv.java +++ b/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosEnv.java @@ -29,7 +29,7 @@ public static Properties getProperties(String testName) { properties.setProperty(DatabaseConfig.PASSWORD, password); properties.setProperty(DatabaseConfig.STORAGE, "cosmos"); properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN, "true"); - properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_FILTERING, "false"); + properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_FILTERING, "true"); properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_ORDERING, "false"); if (databasePrefix.isPresent()) { diff --git a/core/src/integration-test/java/com/scalar/db/storage/cosmos/TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithCosmos.java b/core/src/integration-test/java/com/scalar/db/storage/cosmos/TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithCosmos.java new file mode 100644 index 0000000000..bede3ecc1f --- /dev/null +++ b/core/src/integration-test/java/com/scalar/db/storage/cosmos/TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithCosmos.java @@ -0,0 +1,29 @@ +package com.scalar.db.storage.cosmos; + +import com.scalar.db.transaction.consensuscommit.ConsensusCommitConfig; +import com.scalar.db.transaction.consensuscommit.TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestBase; +import java.util.Map; +import java.util.Properties; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +public class TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithCosmos + extends TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestBase { + + @Override + protected Properties getProps1(String testName) { + Properties properties = CosmosEnv.getProperties(testName); + properties.setProperty(ConsensusCommitConfig.ISOLATION_LEVEL, "SERIALIZABLE"); + return properties; + } + + @Override + protected Map getCreationOptions() { + return CosmosEnv.getCreationOptions(); + } + + @Test + @Override + @Disabled("Cross partition scan with ordering is not supported in Cosmos DB") + public void scan_ScanWithOrderingGivenForCommittedRecord_ShouldReturnRecords() {} +} diff --git a/core/src/integration-test/java/com/scalar/db/storage/dynamo/ConsensusCommitCrossPartitionScanIntegrationTestWithDynamo.java b/core/src/integration-test/java/com/scalar/db/storage/dynamo/ConsensusCommitCrossPartitionScanIntegrationTestWithDynamo.java new file mode 100644 index 0000000000..4bd861449b --- /dev/null +++ b/core/src/integration-test/java/com/scalar/db/storage/dynamo/ConsensusCommitCrossPartitionScanIntegrationTestWithDynamo.java @@ -0,0 +1,29 @@ +package com.scalar.db.storage.dynamo; + +import com.scalar.db.transaction.consensuscommit.ConsensusCommitConfig; +import com.scalar.db.transaction.consensuscommit.ConsensusCommitCrossPartitionScanIntegrationTestBase; +import java.util.Map; +import java.util.Properties; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +public class ConsensusCommitCrossPartitionScanIntegrationTestWithDynamo + extends ConsensusCommitCrossPartitionScanIntegrationTestBase { + + @Override + protected Properties getProps(String testName) { + Properties properties = DynamoEnv.getProperties(testName); + properties.setProperty(ConsensusCommitConfig.ISOLATION_LEVEL, "SERIALIZABLE"); + return properties; + } + + @Override + protected Map getCreationOptions() { + return DynamoEnv.getCreationOptions(); + } + + @Test + @Override + @Disabled("Cross partition scan with ordering is not supported in DynamoDB") + public void scan_CrossPartitionScanWithOrderingGivenForCommittedRecord_ShouldReturnRecords() {} +} diff --git a/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoCrossPartitionScanIntegrationTest.java b/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoCrossPartitionScanIntegrationTest.java new file mode 100644 index 0000000000..d0b4c13303 --- /dev/null +++ b/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoCrossPartitionScanIntegrationTest.java @@ -0,0 +1,26 @@ +package com.scalar.db.storage.dynamo; + +import com.scalar.db.api.DistributedStorageCrossPartitionScanIntegrationTestBase; +import java.util.Map; +import java.util.Properties; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +public class DynamoCrossPartitionScanIntegrationTest + extends DistributedStorageCrossPartitionScanIntegrationTestBase { + + @Override + protected Properties getProperties(String testName) { + return DynamoEnv.getProperties(testName); + } + + @Override + protected Map getCreationOptions() { + return DynamoEnv.getCreationOptions(); + } + + @Test + @Override + @Disabled("Cross partition scan with ordering is not supported in DynamoDB") + public void scan_WithOrderingForNonPrimaryColumns_ShouldReturnProperResult() {} +} diff --git a/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoEnv.java b/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoEnv.java index b6440460d6..cf7e719dfb 100644 --- a/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoEnv.java +++ b/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoEnv.java @@ -40,7 +40,7 @@ public static Properties getProperties(String testName) { properties.setProperty(DatabaseConfig.PASSWORD, secretAccessKey); properties.setProperty(DatabaseConfig.STORAGE, "dynamo"); properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN, "true"); - properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_FILTERING, "false"); + properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_FILTERING, "true"); properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_ORDERING, "false"); // Add testName as a metadata namespace suffix diff --git a/core/src/integration-test/java/com/scalar/db/storage/dynamo/TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithDynamo.java b/core/src/integration-test/java/com/scalar/db/storage/dynamo/TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithDynamo.java new file mode 100644 index 0000000000..ca701b9af8 --- /dev/null +++ b/core/src/integration-test/java/com/scalar/db/storage/dynamo/TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithDynamo.java @@ -0,0 +1,29 @@ +package com.scalar.db.storage.dynamo; + +import com.scalar.db.transaction.consensuscommit.ConsensusCommitConfig; +import com.scalar.db.transaction.consensuscommit.TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestBase; +import java.util.Map; +import java.util.Properties; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +public class TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestWithDynamo + extends TwoPhaseConsensusCommitCrossPartitionScanIntegrationTestBase { + + @Override + protected Properties getProps1(String testName) { + Properties properties = DynamoEnv.getProperties(testName); + properties.setProperty(ConsensusCommitConfig.ISOLATION_LEVEL, "SERIALIZABLE"); + return properties; + } + + @Override + protected Map getCreationOptions() { + return DynamoEnv.getCreationOptions(); + } + + @Test + @Override + @Disabled("Cross partition scan with ordering is not supported in DynamoDB") + public void scan_ScanWithOrderingGivenForCommittedRecord_ShouldReturnRecords() {} +} diff --git a/core/src/main/java/com/scalar/db/common/AbstractDistributedStorage.java b/core/src/main/java/com/scalar/db/common/AbstractDistributedStorage.java index b5208268ae..c7c87d2dfb 100644 --- a/core/src/main/java/com/scalar/db/common/AbstractDistributedStorage.java +++ b/core/src/main/java/com/scalar/db/common/AbstractDistributedStorage.java @@ -76,4 +76,12 @@ protected Put copyAndSetTargetToIfNot(Put put) { protected Delete copyAndSetTargetToIfNot(Delete delete) { return ScalarDbUtils.copyAndSetTargetToIfNot(delete, namespace, tableName); } + + protected Get copyAndPrepareForDynamicFiltering(Get get) { + return ScalarDbUtils.copyAndPrepareForDynamicFiltering(get); + } + + protected Scan copyAndPrepareForDynamicFiltering(Scan scan) { + return ScalarDbUtils.copyAndPrepareForDynamicFiltering(scan); + } } diff --git a/core/src/main/java/com/scalar/db/common/FilterableScanner.java b/core/src/main/java/com/scalar/db/common/FilterableScanner.java new file mode 100644 index 0000000000..bec9070807 --- /dev/null +++ b/core/src/main/java/com/scalar/db/common/FilterableScanner.java @@ -0,0 +1,88 @@ +package com.scalar.db.common; + +import com.google.errorprone.annotations.concurrent.LazyInit; +import com.scalar.db.api.Result; +import com.scalar.db.api.Scan; +import com.scalar.db.api.Scanner; +import com.scalar.db.api.Selection; +import com.scalar.db.api.Selection.Conjunction; +import com.scalar.db.exception.storage.ExecutionException; +import com.scalar.db.util.ScalarDbUtils; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import javax.annotation.concurrent.NotThreadSafe; + +@NotThreadSafe +public class FilterableScanner implements Scanner { + + private final Scanner scanner; + private final List projections; + private final Set conjunctions; + @Nullable private Integer left = null; + @LazyInit private ScannerIterator scannerIterator; + + public FilterableScanner(Selection selection, Scanner scanner) { + this.scanner = scanner; + this.projections = selection.getProjections(); + this.conjunctions = selection.getConjunctions(); + if (selection instanceof Scan) { + Scan scan = (Scan) selection; + this.left = scan.getLimit() > 0 ? scan.getLimit() : null; + } + } + + @SuppressFBWarnings("DLS_DEAD_LOCAL_STORE") + @Override + public Optional one() throws ExecutionException { + if (left != null && left == 0) { + return Optional.empty(); + } + while (true) { + Optional one = scanner.one(); + if (one.isPresent()) { + if (ScalarDbUtils.columnsMatchAnyOfConjunctions(one.get().getColumns(), conjunctions)) { + if (left != null) { + left--; + } + return Optional.of(new ProjectedResult(one.get(), projections)); + } + } else { + return Optional.empty(); + } + } + } + + @Override + public List all() throws ExecutionException { + List ret = new ArrayList<>(); + while (true) { + Optional one = one(); + if (!one.isPresent()) { + break; + } + ret.add(one.get()); + } + return ret; + } + + @Override + @Nonnull + public Iterator iterator() { + if (scannerIterator == null) { + scannerIterator = new ScannerIterator(this); + } + return scannerIterator; + } + + @Override + public void close() throws IOException { + scanner.close(); + } +} diff --git a/core/src/main/java/com/scalar/db/common/ProjectedResult.java b/core/src/main/java/com/scalar/db/common/ProjectedResult.java new file mode 100644 index 0000000000..d9f79eea2e --- /dev/null +++ b/core/src/main/java/com/scalar/db/common/ProjectedResult.java @@ -0,0 +1,143 @@ +package com.scalar.db.common; + +import com.google.common.collect.ImmutableSet; +import com.scalar.db.api.Result; +import com.scalar.db.common.error.CoreError; +import com.scalar.db.io.Column; +import com.scalar.db.io.Key; +import com.scalar.db.io.Value; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import javax.annotation.Nullable; +import javax.annotation.concurrent.Immutable; + +/** An implementation of {@code Result} that only includes projected columns. */ +@Immutable +public class ProjectedResult extends AbstractResult { + + private final Result original; + private final ImmutableSet containedColumnNames; + + public ProjectedResult(Result original, List projections) { + this.original = Objects.requireNonNull(original); + + ImmutableSet.Builder builder = ImmutableSet.builder(); + original.getContainedColumnNames().stream() + .filter(c -> projections.isEmpty() || projections.contains(c)) + .forEach(builder::add); + containedColumnNames = builder.build(); + } + + /** @deprecated As of release 3.8.0. Will be removed in release 5.0.0 */ + @Deprecated + @Override + public Optional getPartitionKey() { + return getKey(original.getPartitionKey()); + } + + /** @deprecated As of release 3.8.0. Will be removed in release 5.0.0 */ + @Deprecated + @Override + public Optional getClusteringKey() { + return getKey(original.getClusteringKey()); + } + + private Optional getKey(Optional key) { + if (!key.isPresent()) { + return Optional.empty(); + } + for (Value value : key.get()) { + if (!containedColumnNames.contains(value.getName())) { + throw new IllegalStateException(CoreError.COLUMN_NOT_FOUND.buildMessage(value.getName())); + } + } + return key; + } + + @Override + public boolean isNull(String columnName) { + checkIfExists(columnName); + return original.isNull(columnName); + } + + @Override + public boolean getBoolean(String columnName) { + checkIfExists(columnName); + return original.getBoolean(columnName); + } + + @Override + public int getInt(String columnName) { + checkIfExists(columnName); + return original.getInt(columnName); + } + + @Override + public long getBigInt(String columnName) { + checkIfExists(columnName); + return original.getBigInt(columnName); + } + + @Override + public float getFloat(String columnName) { + checkIfExists(columnName); + return original.getFloat(columnName); + } + + @Override + public double getDouble(String columnName) { + checkIfExists(columnName); + return original.getDouble(columnName); + } + + @Nullable + @Override + public String getText(String columnName) { + checkIfExists(columnName); + return original.getText(columnName); + } + + @Nullable + @Override + public ByteBuffer getBlobAsByteBuffer(String columnName) { + checkIfExists(columnName); + return original.getBlobAsByteBuffer(columnName); + } + + @Nullable + @Override + public byte[] getBlobAsBytes(String columnName) { + checkIfExists(columnName); + return original.getBlobAsBytes(columnName); + } + + @Nullable + @Override + public Object getAsObject(String columnName) { + checkIfExists(columnName); + return original.getAsObject(columnName); + } + + @Override + public boolean contains(String columnName) { + return containedColumnNames.contains(columnName); + } + + @Override + public Set getContainedColumnNames() { + return containedColumnNames; + } + + @Override + public Map> getColumns() { + return original.getColumns().entrySet().stream() + .filter(e -> containedColumnNames.contains(e.getKey())) + .collect(Collectors.toMap(Entry::getKey, Entry::getValue)); + } +} diff --git a/core/src/main/java/com/scalar/db/common/ResultImpl.java b/core/src/main/java/com/scalar/db/common/ResultImpl.java index eb1ee391ed..10cd8a055a 100644 --- a/core/src/main/java/com/scalar/db/common/ResultImpl.java +++ b/core/src/main/java/com/scalar/db/common/ResultImpl.java @@ -2,6 +2,7 @@ import com.google.common.collect.ImmutableMap; import com.scalar.db.api.TableMetadata; +import com.scalar.db.common.error.CoreError; import com.scalar.db.io.Column; import com.scalar.db.io.Key; import java.nio.ByteBuffer; @@ -12,12 +13,9 @@ import java.util.Set; import javax.annotation.Nullable; import javax.annotation.concurrent.Immutable; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; @Immutable public class ResultImpl extends AbstractResult { - private static final Logger logger = LoggerFactory.getLogger(ResultImpl.class); private final ImmutableMap> columns; private final TableMetadata metadata; @@ -49,8 +47,7 @@ private Optional getKey(LinkedHashSet names) { for (String name : names) { Column column = columns.get(name); if (column == null) { - logger.warn("Full key doesn't seem to be projected into the result"); - return Optional.empty(); + throw new IllegalStateException(CoreError.COLUMN_NOT_FOUND.buildMessage(name)); } builder.add(column); } diff --git a/core/src/main/java/com/scalar/db/common/error/CoreError.java b/core/src/main/java/com/scalar/db/common/error/CoreError.java index 6856b40035..a0af7dfc8e 100644 --- a/core/src/main/java/com/scalar/db/common/error/CoreError.java +++ b/core/src/main/java/com/scalar/db/common/error/CoreError.java @@ -301,12 +301,6 @@ public enum CoreError implements ScalarDbError { STORAGE_NOT_FOUND(Category.USER_ERROR, "0065", "Storage '%s' is not found", "", ""), TRANSACTION_MANAGER_NOT_FOUND( Category.USER_ERROR, "0066", "Transaction manager '%s' is not found", "", ""), - CASSANDRA_CROSS_PARTITION_SCAN_WITH_FILTERING_OR_ORDERING_NOT_SUPPORTED( - Category.USER_ERROR, - "0067", - "Cross-partition scan with filtering or ordering is not supported in Cassandra", - "", - ""), GET_OPERATION_USED_FOR_NON_EXACT_MATCH_SELECTION( Category.USER_ERROR, "0068", @@ -327,12 +321,6 @@ public enum CoreError implements ScalarDbError { "The property 'scalar.db.contact_port' must be greater than or equal to zero", "", ""), - COSMOS_CROSS_PARTITION_SCAN_WITH_FILTERING_OR_ORDERING_NOT_SUPPORTED( - Category.USER_ERROR, - "0072", - "Cross-partition scan with filtering or ordering is not supported in Cosmos DB", - "", - ""), COSMOS_CLUSTERING_KEY_BLOB_TYPE_NOT_SUPPORTED( Category.USER_ERROR, "0073", @@ -371,12 +359,6 @@ public enum CoreError implements ScalarDbError { ""), DYNAMO_ENCODER_CANNOT_ENCODE_TEXT_VALUE_CONTAINING_0X0000( Category.USER_ERROR, "0079", "Cannot encode a Text value that contains '\\u0000'", "", ""), - DYNAMO_CROSS_PARTITION_SCAN_WITH_FILTERING_OR_ORDERING_NOT_SUPPORTED( - Category.USER_ERROR, - "0080", - "Cross-partition scan with filtering or ordering is not supported in DynamoDB", - "", - ""), DYNAMO_INDEX_COLUMN_CANNOT_BE_SET_TO_NULL_OR_EMPTY( Category.USER_ERROR, "0081", @@ -584,6 +566,24 @@ public enum CoreError implements ScalarDbError { "This condition is not allowed for the UpdateIf operation. Condition: %s", "", ""), + CASSANDRA_CROSS_PARTITION_SCAN_WITH_ORDERING_NOT_SUPPORTED( + Category.USER_ERROR, + "0128", + "Cross-partition scan with ordering is not supported in Cassandra", + "", + ""), + COSMOS_CROSS_PARTITION_SCAN_WITH_ORDERING_NOT_SUPPORTED( + Category.USER_ERROR, + "0129", + "Cross-partition scan with ordering is not supported in Cosmos DB", + "", + ""), + DYNAMO_CROSS_PARTITION_SCAN_WITH_ORDERING_NOT_SUPPORTED( + Category.USER_ERROR, + "0130", + "Cross-partition scan with ordering is not supported in DynamoDB", + "", + ""), // // Errors for the concurrency error category diff --git a/core/src/main/java/com/scalar/db/storage/cassandra/Cassandra.java b/core/src/main/java/com/scalar/db/storage/cassandra/Cassandra.java index b6d3ee888b..5be750ee74 100644 --- a/core/src/main/java/com/scalar/db/storage/cassandra/Cassandra.java +++ b/core/src/main/java/com/scalar/db/storage/cassandra/Cassandra.java @@ -2,8 +2,6 @@ import static com.google.common.base.Preconditions.checkArgument; -import com.datastax.driver.core.ResultSet; -import com.datastax.driver.core.Row; import com.datastax.driver.core.Session; import com.google.common.annotations.VisibleForTesting; import com.google.inject.Inject; @@ -16,11 +14,13 @@ import com.scalar.db.api.Scan; import com.scalar.db.api.Scanner; import com.scalar.db.common.AbstractDistributedStorage; +import com.scalar.db.common.FilterableScanner; import com.scalar.db.common.TableMetadataManager; import com.scalar.db.common.checker.OperationChecker; import com.scalar.db.common.error.CoreError; import com.scalar.db.config.DatabaseConfig; import com.scalar.db.exception.storage.ExecutionException; +import java.io.IOException; import java.util.List; import java.util.Optional; import javax.annotation.Nonnull; @@ -46,11 +46,9 @@ public class Cassandra extends AbstractDistributedStorage { public Cassandra(DatabaseConfig config) { super(config); - if (config.isCrossPartitionScanFilteringEnabled() - || config.isCrossPartitionScanOrderingEnabled()) { + if (config.isCrossPartitionScanOrderingEnabled()) { throw new IllegalArgumentException( - CoreError.CASSANDRA_CROSS_PARTITION_SCAN_WITH_FILTERING_OR_ORDERING_NOT_SUPPORTED - .buildMessage()); + CoreError.CASSANDRA_CROSS_PARTITION_SCAN_WITH_ORDERING_NOT_SUPPORTED.buildMessage()); } clusterManager = new ClusterManager(config); @@ -100,19 +98,34 @@ public Optional get(Get get) throws ExecutionException { get = copyAndSetTargetToIfNot(get); operationChecker.check(get); - ResultSet resultSet = handlers.select().handle(get); - Row row = resultSet.one(); - if (row == null) { - return Optional.empty(); - } - Row next = resultSet.one(); - if (next != null) { - throw new IllegalArgumentException( - CoreError.GET_OPERATION_USED_FOR_NON_EXACT_MATCH_SELECTION.buildMessage(get)); + Scanner scanner = null; + try { + if (get.getConjunctions().isEmpty()) { + scanner = getInternal(get); + } else { + scanner = new FilterableScanner(get, getInternal(copyAndPrepareForDynamicFiltering(get))); + } + Optional ret = scanner.one(); + if (scanner.one().isPresent()) { + throw new IllegalArgumentException( + CoreError.GET_OPERATION_USED_FOR_NON_EXACT_MATCH_SELECTION.buildMessage(get)); + } + return ret; + } finally { + if (scanner != null) { + try { + scanner.close(); + } catch (IOException e) { + logger.warn("Failed to close the scanner", e); + } + } } - return Optional.of( - new ResultInterpreter(get.getProjections(), metadataManager.getTableMetadata(get)) - .interpret(row)); + } + + private Scanner getInternal(Get get) throws ExecutionException { + return new ScannerImpl( + handlers.select().handle(get), + new ResultInterpreter(get.getProjections(), metadataManager.getTableMetadata(get))); } @Override @@ -121,10 +134,16 @@ public Scanner scan(Scan scan) throws ExecutionException { scan = copyAndSetTargetToIfNot(scan); operationChecker.check(scan); - ResultSet results = handlers.select().handle(scan); + if (scan.getConjunctions().isEmpty()) { + return scanInternal(scan); + } else { + return new FilterableScanner(scan, scanInternal(copyAndPrepareForDynamicFiltering(scan))); + } + } + private Scanner scanInternal(Scan scan) throws ExecutionException { return new ScannerImpl( - results, + handlers.select().handle(scan), new ResultInterpreter(scan.getProjections(), metadataManager.getTableMetadata(scan))); } diff --git a/core/src/main/java/com/scalar/db/storage/cosmos/Cosmos.java b/core/src/main/java/com/scalar/db/storage/cosmos/Cosmos.java index 89c8f95de9..7aecec1735 100644 --- a/core/src/main/java/com/scalar/db/storage/cosmos/Cosmos.java +++ b/core/src/main/java/com/scalar/db/storage/cosmos/Cosmos.java @@ -14,11 +14,13 @@ import com.scalar.db.api.Scan; import com.scalar.db.api.Scanner; import com.scalar.db.common.AbstractDistributedStorage; +import com.scalar.db.common.FilterableScanner; import com.scalar.db.common.TableMetadataManager; import com.scalar.db.common.checker.OperationChecker; import com.scalar.db.common.error.CoreError; import com.scalar.db.config.DatabaseConfig; import com.scalar.db.exception.storage.ExecutionException; +import java.io.IOException; import java.util.List; import java.util.Optional; import javax.annotation.Nonnull; @@ -46,11 +48,9 @@ public class Cosmos extends AbstractDistributedStorage { public Cosmos(DatabaseConfig databaseConfig) { super(databaseConfig); - if (databaseConfig.isCrossPartitionScanFilteringEnabled() - || databaseConfig.isCrossPartitionScanOrderingEnabled()) { + if (databaseConfig.isCrossPartitionScanOrderingEnabled()) { throw new IllegalArgumentException( - CoreError.COSMOS_CROSS_PARTITION_SCAN_WITH_FILTERING_OR_ORDERING_NOT_SUPPORTED - .buildMessage()); + CoreError.COSMOS_CROSS_PARTITION_SCAN_WITH_ORDERING_NOT_SUPPORTED.buildMessage()); } CosmosConfig config = new CosmosConfig(databaseConfig); @@ -98,14 +98,30 @@ public Optional get(Get get) throws ExecutionException { get = copyAndSetTargetToIfNot(get); operationChecker.check(get); - Scanner scanner = selectStatementHandler.handle(get); - Optional ret = scanner.one(); - if (scanner.one().isPresent()) { - throw new IllegalArgumentException( - CoreError.GET_OPERATION_USED_FOR_NON_EXACT_MATCH_SELECTION.buildMessage(get)); + Scanner scanner = null; + try { + if (get.getConjunctions().isEmpty()) { + scanner = selectStatementHandler.handle(get); + } else { + scanner = + new FilterableScanner( + get, selectStatementHandler.handle(copyAndPrepareForDynamicFiltering(get))); + } + Optional ret = scanner.one(); + if (scanner.one().isPresent()) { + throw new IllegalArgumentException( + CoreError.GET_OPERATION_USED_FOR_NON_EXACT_MATCH_SELECTION.buildMessage(get)); + } + return ret; + } finally { + if (scanner != null) { + try { + scanner.close(); + } catch (IOException e) { + logger.warn("Failed to close the scanner", e); + } + } } - - return ret; } @Override @@ -113,7 +129,12 @@ public Scanner scan(Scan scan) throws ExecutionException { scan = copyAndSetTargetToIfNot(scan); operationChecker.check(scan); - return selectStatementHandler.handle(scan); + if (scan.getConjunctions().isEmpty()) { + return selectStatementHandler.handle(scan); + } else { + return new FilterableScanner( + scan, selectStatementHandler.handle(copyAndPrepareForDynamicFiltering(scan))); + } } @Override diff --git a/core/src/main/java/com/scalar/db/storage/dynamo/Dynamo.java b/core/src/main/java/com/scalar/db/storage/dynamo/Dynamo.java index bc79e2e276..e841d59e00 100644 --- a/core/src/main/java/com/scalar/db/storage/dynamo/Dynamo.java +++ b/core/src/main/java/com/scalar/db/storage/dynamo/Dynamo.java @@ -13,6 +13,7 @@ import com.scalar.db.api.Scan; import com.scalar.db.api.Scanner; import com.scalar.db.common.AbstractDistributedStorage; +import com.scalar.db.common.FilterableScanner; import com.scalar.db.common.TableMetadataManager; import com.scalar.db.common.checker.OperationChecker; import com.scalar.db.common.error.CoreError; @@ -52,11 +53,9 @@ public class Dynamo extends AbstractDistributedStorage { public Dynamo(DatabaseConfig databaseConfig) { super(databaseConfig); - if (databaseConfig.isCrossPartitionScanFilteringEnabled() - || databaseConfig.isCrossPartitionScanOrderingEnabled()) { + if (databaseConfig.isCrossPartitionScanOrderingEnabled()) { throw new IllegalArgumentException( - CoreError.DYNAMO_CROSS_PARTITION_SCAN_WITH_FILTERING_OR_ORDERING_NOT_SUPPORTED - .buildMessage()); + CoreError.DYNAMO_CROSS_PARTITION_SCAN_WITH_ORDERING_NOT_SUPPORTED.buildMessage()); } DynamoConfig config = new DynamoConfig(databaseConfig); @@ -118,7 +117,13 @@ public Optional get(Get get) throws ExecutionException { Scanner scanner = null; try { - scanner = selectStatementHandler.handle(get); + if (get.getConjunctions().isEmpty()) { + scanner = selectStatementHandler.handle(get); + } else { + scanner = + new FilterableScanner( + get, selectStatementHandler.handle(copyAndPrepareForDynamicFiltering(get))); + } Optional ret = scanner.one(); if (scanner.one().isPresent()) { throw new IllegalArgumentException( @@ -141,7 +146,12 @@ public Scanner scan(Scan scan) throws ExecutionException { scan = copyAndSetTargetToIfNot(scan); operationChecker.check(scan); - return selectStatementHandler.handle(scan); + if (scan.getConjunctions().isEmpty()) { + return selectStatementHandler.handle(scan); + } else { + return new FilterableScanner( + scan, selectStatementHandler.handle(copyAndPrepareForDynamicFiltering(scan))); + } } @Override diff --git a/core/src/main/java/com/scalar/db/transaction/consensuscommit/FilteredResult.java b/core/src/main/java/com/scalar/db/transaction/consensuscommit/FilteredResult.java index 90278fc707..a03a2b0ef9 100644 --- a/core/src/main/java/com/scalar/db/transaction/consensuscommit/FilteredResult.java +++ b/core/src/main/java/com/scalar/db/transaction/consensuscommit/FilteredResult.java @@ -5,6 +5,7 @@ import com.scalar.db.api.Result; import com.scalar.db.api.TableMetadata; import com.scalar.db.common.AbstractResult; +import com.scalar.db.common.error.CoreError; import com.scalar.db.io.Column; import com.scalar.db.io.Key; import com.scalar.db.io.Value; @@ -18,15 +19,12 @@ import java.util.stream.Collectors; import javax.annotation.Nullable; import javax.annotation.concurrent.Immutable; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * An implementation of {@code Result} to filter out unprojected columns and transaction columns. */ @Immutable public class FilteredResult extends AbstractResult { - private static final Logger logger = LoggerFactory.getLogger(FilteredResult.class); private final Result original; private final ImmutableSet containedColumnNames; @@ -69,8 +67,7 @@ private Optional getKey(Optional key) { } for (Value value : key.get()) { if (!containedColumnNames.contains(value.getName())) { - logger.warn("Full key doesn't seem to be projected into the result"); - return Optional.empty(); + throw new IllegalStateException(CoreError.COLUMN_NOT_FOUND.buildMessage(value.getName())); } } return key; diff --git a/core/src/main/java/com/scalar/db/transaction/consensuscommit/Snapshot.java b/core/src/main/java/com/scalar/db/transaction/consensuscommit/Snapshot.java index 3de676c716..8cadff5226 100644 --- a/core/src/main/java/com/scalar/db/transaction/consensuscommit/Snapshot.java +++ b/core/src/main/java/com/scalar/db/transaction/consensuscommit/Snapshot.java @@ -3,13 +3,10 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.MoreObjects; import com.google.common.collect.ComparisonChain; -import com.scalar.db.api.ConditionalExpression; -import com.scalar.db.api.ConditionalExpression.Operator; import com.scalar.db.api.Consistency; import com.scalar.db.api.Delete; import com.scalar.db.api.DistributedStorage; import com.scalar.db.api.Get; -import com.scalar.db.api.LikeExpression; import com.scalar.db.api.Operation; import com.scalar.db.api.Put; import com.scalar.db.api.Result; @@ -17,7 +14,6 @@ import com.scalar.db.api.ScanAll; import com.scalar.db.api.ScanWithIndex; import com.scalar.db.api.Scanner; -import com.scalar.db.api.Selection.Conjunction; import com.scalar.db.api.TableMetadata; import com.scalar.db.common.error.CoreError; import com.scalar.db.exception.storage.ExecutionException; @@ -40,9 +36,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.regex.Pattern; import javax.annotation.Nonnull; -import javax.annotation.Nullable; import javax.annotation.concurrent.Immutable; import javax.annotation.concurrent.NotThreadSafe; import org.slf4j.Logger; @@ -345,18 +339,8 @@ private boolean isWriteSetOverlappedWith(ScanAll scan) { } Map> columns = getAllColumns(put); - for (Conjunction conjunction : scan.getConjunctions()) { - boolean allMatched = true; - for (ConditionalExpression condition : conjunction.getConditions()) { - if (!columns.containsKey(condition.getColumn().getName()) - || !match(columns.get(condition.getColumn().getName()), condition)) { - allMatched = false; - break; - } - } - if (allMatched) { - return true; - } + if (ScalarDbUtils.columnsMatchAnyOfConjunctions(columns, scan.getConjunctions())) { + return true; } } return false; @@ -371,32 +355,6 @@ private Map> getAllColumns(Put put) { return columns; } - @SuppressWarnings("unchecked") - private boolean match(Column column, ConditionalExpression condition) { - assert column.getClass() == condition.getColumn().getClass(); - switch (condition.getOperator()) { - case EQ: - case IS_NULL: - return column.equals(condition.getColumn()); - case NE: - case IS_NOT_NULL: - return !column.equals(condition.getColumn()); - case GT: - return column.compareTo((Column) condition.getColumn()) > 0; - case GTE: - return column.compareTo((Column) condition.getColumn()) >= 0; - case LT: - return column.compareTo((Column) condition.getColumn()) < 0; - case LTE: - return column.compareTo((Column) condition.getColumn()) <= 0; - case LIKE: - case NOT_LIKE: - return isMatched((LikeExpression) condition, column.getTextValue()); - default: - throw new AssertionError("Unknown operator: " + condition.getOperator()); - } - } - @VisibleForTesting void toSerializableWithExtraWrite(MutationComposer composer) throws ExecutionException, PreparationConflictException { @@ -567,61 +525,6 @@ public boolean isValidationRequired() { return isExtraReadEnabled(); } - @VisibleForTesting - boolean isMatched(LikeExpression likeExpression, String value) { - String escape = likeExpression.getEscape(); - String regexPattern = - convertRegexPatternFrom( - likeExpression.getTextValue(), escape.isEmpty() ? null : escape.charAt(0)); - if (likeExpression.getOperator().equals(Operator.LIKE)) { - return value != null && Pattern.compile(regexPattern).matcher(value).matches(); - } else { - return value != null && !Pattern.compile(regexPattern).matcher(value).matches(); - } - } - - /** - * Convert SQL 'like' pattern to a Java regular expression. Underscores (_) are converted to '.' - * and percent signs (%) are converted to '.*', other characters are quoted literally. If an - * escape character specified, escaping is done for '_', '%', and the escape character itself. - * Although we validate the pattern when constructing {@code LikeExpression}, we will assert it - * just in case. This method is implemented referencing the following Spark SQL implementation. - * https://github.com/apache/spark/blob/a8eadebd686caa110c4077f4199d11e797146dc5/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/StringUtils.scala - * - * @param likePattern a SQL LIKE pattern to convert - * @param escape an escape character. - * @return the equivalent Java regular expression of the given pattern - */ - private String convertRegexPatternFrom(String likePattern, @Nullable Character escape) { - assert likePattern != null : "LIKE pattern must not be null"; - - StringBuilder out = new StringBuilder(); - char[] chars = likePattern.toCharArray(); - for (int i = 0; i < chars.length; i++) { - char c = chars[i]; - if (escape != null && c == escape && i + 1 < chars.length) { - char nextChar = chars[++i]; - if (nextChar == '_' || nextChar == '%') { - out.append(Pattern.quote(Character.toString(nextChar))); - } else if (nextChar == escape) { - out.append(Pattern.quote(Character.toString(nextChar))); - } else { - throw new AssertionError("LIKE pattern must not include only escape character"); - } - } else if (escape != null && c == escape) { - throw new AssertionError("LIKE pattern must not end with escape character"); - } else if (c == '_') { - out.append("."); - } else if (c == '%') { - out.append(".*"); - } else { - out.append(Pattern.quote(Character.toString(c))); - } - } - - return "(?s)" + out; // (?s) enables dotall mode, causing "." to match new lines - } - @Immutable public static final class Key implements Comparable { private final String namespace; diff --git a/core/src/main/java/com/scalar/db/util/ScalarDbUtils.java b/core/src/main/java/com/scalar/db/util/ScalarDbUtils.java index a6e341f66e..cdd5eee6a1 100644 --- a/core/src/main/java/com/scalar/db/util/ScalarDbUtils.java +++ b/core/src/main/java/com/scalar/db/util/ScalarDbUtils.java @@ -1,10 +1,14 @@ package com.scalar.db.util; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Streams; +import com.scalar.db.api.ConditionalExpression; +import com.scalar.db.api.ConditionalExpression.Operator; import com.scalar.db.api.Delete; import com.scalar.db.api.Get; import com.scalar.db.api.GetWithIndex; import com.scalar.db.api.Insert; +import com.scalar.db.api.LikeExpression; import com.scalar.db.api.Mutation; import com.scalar.db.api.Operation; import com.scalar.db.api.Put; @@ -12,6 +16,7 @@ import com.scalar.db.api.ScanAll; import com.scalar.db.api.ScanWithIndex; import com.scalar.db.api.Selection; +import com.scalar.db.api.Selection.Conjunction; import com.scalar.db.api.TableMetadata; import com.scalar.db.api.Update; import com.scalar.db.api.UpdateIf; @@ -34,13 +39,18 @@ import com.scalar.db.io.TextColumn; import com.scalar.db.io.TextValue; import com.scalar.db.io.Value; +import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletionService; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; import java.util.stream.Collectors; +import javax.annotation.Nullable; public final class ScalarDbUtils { @@ -284,4 +294,138 @@ public static void checkUpdate(Update update) { } }); } + + public static Get copyAndPrepareForDynamicFiltering(Get get) { + Get ret = Get.newBuilder(get).build(); // copy + if (!ret.getProjections().isEmpty()) { + // Add columns in conditions into projections to use them in dynamic filtering + ScalarDbUtils.getColumnNamesUsedIn(ret.getConjunctions()).stream() + .filter(columnName -> !ret.getProjections().contains(columnName)) + .forEach(ret::withProjection); + } + return ret; + } + + public static Scan copyAndPrepareForDynamicFiltering(Scan scan) { + // Ignore limit to control it during dynamic filtering + Scan ret = Scan.newBuilder(scan).limit(0).build(); // copy + if (!ret.getProjections().isEmpty()) { + // Add columns in conditions into projections to use them in dynamic filtering + ScalarDbUtils.getColumnNamesUsedIn(ret.getConjunctions()).stream() + .filter(columnName -> !ret.getProjections().contains(columnName)) + .forEach(ret::withProjection); + } + return ret; + } + + public static Set getColumnNamesUsedIn(Set conjunctions) { + Set columns = new HashSet<>(); + conjunctions.forEach( + conjunction -> + conjunction + .getConditions() + .forEach(condition -> columns.add(condition.getColumn().getName()))); + return columns; + } + + public static boolean columnsMatchAnyOfConjunctions( + Map> columns, Set conjunctions) { + for (Conjunction conjunction : conjunctions) { + boolean allMatched = true; + for (ConditionalExpression condition : conjunction.getConditions()) { + if (!columns.containsKey(condition.getColumn().getName()) + || !columnMatchesCondition(columns.get(condition.getColumn().getName()), condition)) { + allMatched = false; + break; + } + } + if (allMatched) { + return true; + } + } + return false; + } + + @SuppressWarnings("unchecked") + private static boolean columnMatchesCondition( + Column column, ConditionalExpression condition) { + assert column.getClass() == condition.getColumn().getClass(); + switch (condition.getOperator()) { + case EQ: + case IS_NULL: + return column.equals(condition.getColumn()); + case NE: + case IS_NOT_NULL: + return !column.equals(condition.getColumn()); + case GT: + return column.compareTo((Column) condition.getColumn()) > 0; + case GTE: + return column.compareTo((Column) condition.getColumn()) >= 0; + case LT: + return column.compareTo((Column) condition.getColumn()) < 0; + case LTE: + return column.compareTo((Column) condition.getColumn()) <= 0; + case LIKE: + case NOT_LIKE: + // assert condition instanceof LikeExpression; + return stringMatchesLikeExpression(column.getTextValue(), (LikeExpression) condition); + default: + throw new AssertionError("Unknown operator: " + condition.getOperator()); + } + } + + @VisibleForTesting + static boolean stringMatchesLikeExpression(String value, LikeExpression likeExpression) { + String escape = likeExpression.getEscape(); + String regexPattern = + convertRegexPatternFrom( + likeExpression.getTextValue(), escape.isEmpty() ? null : escape.charAt(0)); + if (likeExpression.getOperator().equals(Operator.LIKE)) { + return value != null && Pattern.compile(regexPattern).matcher(value).matches(); + } else { + return value != null && !Pattern.compile(regexPattern).matcher(value).matches(); + } + } + + /** + * Convert SQL 'like' pattern to a Java regular expression. Underscores (_) are converted to '.' + * and percent signs (%) are converted to '.*', other characters are quoted literally. If an + * escape character specified, escaping is done for '_', '%', and the escape character itself. + * Although we validate the pattern when constructing {@code LikeExpression}, we will assert it + * just in case. This method is implemented referencing the following Spark SQL implementation. + * https://github.com/apache/spark/blob/a8eadebd686caa110c4077f4199d11e797146dc5/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/StringUtils.scala + * + * @param likePattern a SQL LIKE pattern to convert + * @param escape an escape character. + * @return the equivalent Java regular expression of the given pattern + */ + private static String convertRegexPatternFrom(String likePattern, @Nullable Character escape) { + assert likePattern != null : "LIKE pattern must not be null"; + + StringBuilder out = new StringBuilder(); + char[] chars = likePattern.toCharArray(); + for (int i = 0; i < chars.length; i++) { + char c = chars[i]; + if (escape != null && c == escape && i + 1 < chars.length) { + char nextChar = chars[++i]; + if (nextChar == '_' || nextChar == '%') { + out.append(Pattern.quote(Character.toString(nextChar))); + } else if (nextChar == escape) { + out.append(Pattern.quote(Character.toString(nextChar))); + } else { + throw new AssertionError("LIKE pattern must not include only escape character"); + } + } else if (escape != null && c == escape) { + throw new AssertionError("LIKE pattern must not end with escape character"); + } else if (c == '_') { + out.append("."); + } else if (c == '%') { + out.append(".*"); + } else { + out.append(Pattern.quote(Character.toString(c))); + } + } + + return "(?s)" + out; // (?s) enables dotall mode, causing "." to match new lines + } } diff --git a/core/src/test/java/com/scalar/db/common/FilterableScannerTest.java b/core/src/test/java/com/scalar/db/common/FilterableScannerTest.java new file mode 100644 index 0000000000..56ee226eee --- /dev/null +++ b/core/src/test/java/com/scalar/db/common/FilterableScannerTest.java @@ -0,0 +1,137 @@ +package com.scalar.db.common; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; +import com.scalar.db.api.ConditionBuilder; +import com.scalar.db.api.Result; +import com.scalar.db.api.Scan; +import com.scalar.db.api.Scanner; +import com.scalar.db.api.Selection.Conjunction; +import com.scalar.db.exception.storage.ExecutionException; +import com.scalar.db.io.IntColumn; +import java.util.Iterator; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.Optional; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +public class FilterableScannerTest { + + @Mock private Scan scan; + @Mock private Scanner scanner; + @Mock private Result result1, result2, result3; + + @BeforeEach + public void setUp() throws Exception { + MockitoAnnotations.openMocks(this).close(); + + when(scanner.one()) + .thenReturn(Optional.of(result1)) + .thenReturn(Optional.of(result2)) + .thenReturn(Optional.of(result3)) + .thenReturn(Optional.empty()); + when(result1.getColumns()).thenReturn(ImmutableMap.of("col", IntColumn.of("col", 0))); + when(result2.getColumns()).thenReturn(ImmutableMap.of("col", IntColumn.of("col", 1))); + when(result3.getColumns()).thenReturn(ImmutableMap.of("col", IntColumn.of("col", 2))); + when(scan.getConjunctions()) + .thenReturn( + ImmutableSet.of(Conjunction.of(ConditionBuilder.column("col").isGreaterThanInt(0)))); + } + + @Test + public void one_ShouldReturnResult() throws ExecutionException { + // Arrange + FilterableScanner filterableScanner = new FilterableScanner(scan, scanner); + + // Act + Optional actual1 = filterableScanner.one(); + Optional actual2 = filterableScanner.one(); + Optional actual3 = filterableScanner.one(); + + // Assert + assertThat(actual1).isPresent(); + assertThat(actual1.get()).isEqualTo(result2); + assertThat(actual2).isPresent(); + assertThat(actual2.get()).isEqualTo(result3); + assertThat(actual3).isNotPresent(); + verify(scanner, times(4)).one(); + } + + @Test + public void one_AfterExceedingLimit_ShouldReturnEmpty() throws ExecutionException { + // Arrange + when(scan.getLimit()).thenReturn(1); + FilterableScanner filterableScanner = new FilterableScanner(scan, scanner); + + // Act + Optional actual1 = filterableScanner.one(); + Optional actual2 = filterableScanner.one(); + + // Assert + assertThat(actual1).isPresent(); + assertThat(actual1.get()).isEqualTo(result2); + assertThat(actual2).isNotPresent(); + verify(scanner, times(2)).one(); + } + + @Test + public void all_ShouldReturnResults() throws ExecutionException { + // Arrange + FilterableScanner filterableScanner = new FilterableScanner(scan, scanner); + + // Act + List results1 = filterableScanner.all(); + List results2 = filterableScanner.all(); + + // Assert + assertThat(results1.size()).isEqualTo(2); + assertThat(results1.get(0)).isEqualTo(result2); + assertThat(results1.get(1)).isEqualTo(result3); + assertThat(results2).isEmpty(); + verify(scanner, times(5)).one(); + } + + @Test + public void all_WithLimit_ShouldReturnLimitedResults() throws ExecutionException { + // Arrange + when(scan.getLimit()).thenReturn(1); + FilterableScanner filterableScanner = new FilterableScanner(scan, scanner); + + // Act + List results1 = filterableScanner.all(); + List results2 = filterableScanner.all(); + + // Assert + assertThat(results1.size()).isEqualTo(1); + assertThat(results1.get(0)).isEqualTo(result2); + assertThat(results2).isEmpty(); + verify(scanner, times(2)).one(); + } + + @Test + public void iterator_ShouldReturnResults() throws ExecutionException { + // Arrange + FilterableScanner filterableScanner = new FilterableScanner(scan, scanner); + + // Act + Iterator iterator = filterableScanner.iterator(); + + // Assert + assertThat(iterator.hasNext()).isTrue(); + assertThat(iterator.next()).isEqualTo(result2); + assertThat(iterator.hasNext()).isTrue(); + assertThat(iterator.next()).isEqualTo(result3); + assertThat(iterator.hasNext()).isFalse(); + assertThatThrownBy(iterator::next).isInstanceOf(NoSuchElementException.class); + verify(scanner, times(5)).one(); + } +} diff --git a/core/src/test/java/com/scalar/db/common/ProjectedResultTest.java b/core/src/test/java/com/scalar/db/common/ProjectedResultTest.java new file mode 100644 index 0000000000..fe6be6e50c --- /dev/null +++ b/core/src/test/java/com/scalar/db/common/ProjectedResultTest.java @@ -0,0 +1,278 @@ +package com.scalar.db.common; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.catchThrowable; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.scalar.db.api.Result; +import com.scalar.db.api.TableMetadata; +import com.scalar.db.io.Column; +import com.scalar.db.io.DataType; +import com.scalar.db.io.IntColumn; +import com.scalar.db.io.Key; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class ProjectedResultTest { + + private static final String ACCOUNT_ID = "account_id"; + private static final String ACCOUNT_TYPE = "account_type"; + private static final String BALANCE = "balance"; + + private static final TableMetadata TABLE_METADATA = + TableMetadata.newBuilder() + .addColumn(ACCOUNT_ID, DataType.INT) + .addColumn(ACCOUNT_TYPE, DataType.INT) + .addColumn(BALANCE, DataType.INT) + .addPartitionKey(ACCOUNT_ID) + .addClusteringKey(ACCOUNT_TYPE) + .build(); + + private static final IntColumn ACCOUNT_ID_COLUMN = IntColumn.of(ACCOUNT_ID, 0); + private static final IntColumn ACCOUNT_TYPE_COLUMN = IntColumn.of(ACCOUNT_TYPE, 1); + private static final IntColumn BALANCE_COLUMN = IntColumn.of(BALANCE, 2); + + private Result result; + + @BeforeEach + public void setUp() { + // Arrange + Map> columns = + ImmutableMap.>builder() + .put(ACCOUNT_ID, ACCOUNT_ID_COLUMN) + .put(ACCOUNT_TYPE, ACCOUNT_TYPE_COLUMN) + .put(BALANCE, BALANCE_COLUMN) + .build(); + + result = new ResultImpl(columns, TABLE_METADATA); + } + + @Test + public void getPartitionKey_RequiredValuesGiven_ShouldReturnPartitionKey() { + // Arrange + ProjectedResult projectedResult = new ProjectedResult(result, Collections.emptyList()); + + // Act + Optional key = projectedResult.getPartitionKey(); + + // Assert + assertThat(key.isPresent()).isTrue(); + assertThat(key.get().getColumns().size()).isEqualTo(1); + assertThat(key.get().getIntValue(0)).isEqualTo(0); + } + + @Test + public void getClusteringKey_RequiredValuesGiven_ShouldReturnPartitionKey() { + // Arrange + ProjectedResult projectedResult = new ProjectedResult(result, Collections.emptyList()); + + // Act + Optional key = projectedResult.getClusteringKey(); + + // Assert + assertThat(key.isPresent()).isTrue(); + assertThat(key.get().getColumns().size()).isEqualTo(1); + assertThat(key.get().getIntValue(0)).isEqualTo(1); + } + + @Test + public void getPartitionKey_NotRequiredValuesGiven_ShouldThrowIllegalStateException() { + // Arrange + ProjectedResult projectedResult = new ProjectedResult(result, ImmutableList.of(BALANCE)); + + // Act + Throwable thrown = catchThrowable(projectedResult::getPartitionKey); + + // Assert + assertThat(thrown).isInstanceOf(IllegalStateException.class); + } + + @Test + public void getClusteringKey_NotRequiredValuesGiven_ShouldThrowIllegalStateException() { + // Arrange + ProjectedResult projectedResult = new ProjectedResult(result, ImmutableList.of(BALANCE)); + + // Act + Throwable thrown = catchThrowable(projectedResult::getClusteringKey); + + // Assert + assertThat(thrown).isInstanceOf(IllegalStateException.class); + } + + @Test + public void getClusteringKey_WithoutClusteringKeySchemaGiven_ShouldReturnEmpty() { + // Arrange + Map> columns = + ImmutableMap.>builder() + .put(ACCOUNT_ID, ACCOUNT_ID_COLUMN) + .put(ACCOUNT_TYPE, ACCOUNT_TYPE_COLUMN) + .put(BALANCE, BALANCE_COLUMN) + .build(); + result = + new ResultImpl( + columns, + TableMetadata.newBuilder() + .addColumn(ACCOUNT_ID, DataType.INT) + .addColumn(BALANCE, DataType.INT) + .addPartitionKey(ACCOUNT_ID) + .build()); + ProjectedResult projectedResult = new ProjectedResult(result, ImmutableList.of()); + + // Act + Optional key = projectedResult.getClusteringKey(); + + // Assert + assertThat(key.isPresent()).isFalse(); + } + + @Test + public void withoutProjections_ShouldContainAllColumns() { + // Arrange + ProjectedResult projectedResult = new ProjectedResult(result, Collections.emptyList()); + + // Act Assert + assertThat(projectedResult.getColumns().keySet()) + .containsOnly(ACCOUNT_ID, ACCOUNT_TYPE, BALANCE); + assertThat(projectedResult.getColumns().values()) + .containsOnly(ACCOUNT_ID_COLUMN, ACCOUNT_TYPE_COLUMN, BALANCE_COLUMN); + assertThat(projectedResult.getContainedColumnNames()) + .containsOnly(ACCOUNT_ID, ACCOUNT_TYPE, BALANCE); + + assertThat(projectedResult.contains(ACCOUNT_ID)).isTrue(); + assertThat(projectedResult.isNull(ACCOUNT_ID)).isFalse(); + assertThat(projectedResult.getInt(ACCOUNT_ID)).isEqualTo(0); + assertThat(projectedResult.getAsObject(ACCOUNT_ID)).isEqualTo(0); + + assertThat(projectedResult.contains(ACCOUNT_TYPE)).isTrue(); + assertThat(projectedResult.isNull(ACCOUNT_TYPE)).isFalse(); + assertThat(projectedResult.getInt(ACCOUNT_TYPE)).isEqualTo(1); + assertThat(projectedResult.getAsObject(ACCOUNT_TYPE)).isEqualTo(1); + + assertThat(projectedResult.contains(BALANCE)).isTrue(); + assertThat(projectedResult.isNull(BALANCE)).isFalse(); + assertThat(projectedResult.getInt(BALANCE)).isEqualTo(2); + assertThat(projectedResult.getAsObject(BALANCE)).isEqualTo(2); + } + + @Test + public void withProjections_ShouldContainProjectedColumns() { + // Arrange + ProjectedResult projectedResult = + new ProjectedResult(result, Arrays.asList(ACCOUNT_ID, BALANCE)); + + // Act Assert + assertThat(projectedResult.getColumns().keySet()).containsOnly(ACCOUNT_ID, BALANCE); + assertThat(projectedResult.getColumns().values()) + .containsOnly(ACCOUNT_ID_COLUMN, BALANCE_COLUMN); + assertThat(projectedResult.getContainedColumnNames()) + .isEqualTo(new HashSet<>(Arrays.asList(ACCOUNT_ID, BALANCE))); + + assertThat(projectedResult.contains(ACCOUNT_ID)).isTrue(); + assertThat(projectedResult.isNull(ACCOUNT_ID)).isFalse(); + assertThat(projectedResult.getInt(ACCOUNT_ID)).isEqualTo(0); + assertThat(projectedResult.getAsObject(ACCOUNT_ID)).isEqualTo(0); + + assertThat(projectedResult.contains(ACCOUNT_TYPE)).isFalse(); + + assertThat(projectedResult.contains(BALANCE)).isTrue(); + assertThat(projectedResult.isNull(BALANCE)).isFalse(); + assertThat(projectedResult.getInt(BALANCE)).isEqualTo(2); + assertThat(projectedResult.getAsObject(BALANCE)).isEqualTo(2); + } + + @Test + public void equals_SameResultGiven_WithoutProjections_ShouldReturnTrue() { + // Arrange + ProjectedResult projectedResult = new ProjectedResult(result, Collections.emptyList()); + ProjectedResult anotherFilterResult = new ProjectedResult(result, Collections.emptyList()); + + // Act + boolean isEqual = projectedResult.equals(anotherFilterResult); + + // Assert + assertThat(isEqual).isTrue(); + } + + @Test + public void equals_DifferentResiltGiven_WithoutProjections_ShouldReturnFalse() { + // Arrange + ProjectedResult projectedResult = new ProjectedResult(result, Collections.emptyList()); + ProjectedResult anotherFilterResult = + new ProjectedResult( + new ResultImpl(Collections.emptyMap(), TABLE_METADATA), Collections.emptyList()); + + // Act + boolean isEqual = projectedResult.equals(anotherFilterResult); + + // Assert + assertThat(isEqual).isFalse(); + } + + @Test + public void equals_SameResultGiven_WithProjections_ShouldReturnTrue() { + // Arrange + ProjectedResult projectedResult = + new ProjectedResult(result, Collections.singletonList(BALANCE)); + ProjectedResult anotherFilterResult = + new ProjectedResult(result, Collections.singletonList(BALANCE)); + + // Act + boolean isEqual = projectedResult.equals(anotherFilterResult); + + // Assert + assertThat(isEqual).isTrue(); + } + + @Test + public void equals_ResultImplWithSameValuesGiven_WithoutProjections_ShouldReturnTrue() { + // Arrange + Result projectedResult = new ProjectedResult(result, Collections.emptyList()); + Result anotherResult = + new ResultImpl( + ImmutableMap.of( + ACCOUNT_ID, + ACCOUNT_ID_COLUMN, + ACCOUNT_TYPE, + ACCOUNT_TYPE_COLUMN, + BALANCE, + BALANCE_COLUMN), + TABLE_METADATA); + + // Act + boolean isEqual = projectedResult.equals(anotherResult); + + // Assert + assertThat(isEqual).isTrue(); + } + + @Test + public void equals_ResultImplWithSameValuesGiven_WithProjections_ShouldReturnFalse() { + // Arrange + Result projectedResult = new ProjectedResult(result, Collections.singletonList(BALANCE)); + + // Act + boolean isEqual = projectedResult.equals(result); + + // Assert + assertThat(isEqual).isFalse(); + } + + @Test + public void equals_ResultImplWithDifferentValuesGiven_WithSameProjections_ShouldReturnTrue() { + // Arrange + Result projectedResult = new ProjectedResult(result, Collections.singletonList(BALANCE)); + Result anotherResult = new ResultImpl(ImmutableMap.of(BALANCE, BALANCE_COLUMN), TABLE_METADATA); + + // Act + boolean isEqual = projectedResult.equals(anotherResult); + + // Assert + assertThat(isEqual).isTrue(); + } +} diff --git a/core/src/test/java/com/scalar/db/common/ResultImplTest.java b/core/src/test/java/com/scalar/db/common/ResultImplTest.java index c2757b308e..8f06cdb147 100644 --- a/core/src/test/java/com/scalar/db/common/ResultImplTest.java +++ b/core/src/test/java/com/scalar/db/common/ResultImplTest.java @@ -2,6 +2,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.catchThrowable; import com.google.common.collect.ImmutableMap; import com.scalar.db.api.TableMetadata; @@ -489,7 +490,7 @@ public void getPartitionKey_RequiredValuesGiven_ShouldReturnPartitionKey() { } @Test - public void getPartitionKey_NotRequiredValuesGiven_ShouldReturnPartitionKey() { + public void getPartitionKey_NotRequiredValuesGiven_ShouldThrowIllegalStateException() { // Arrange ResultImpl result = new ResultImpl( @@ -499,10 +500,10 @@ public void getPartitionKey_NotRequiredValuesGiven_ShouldReturnPartitionKey() { TABLE_METADATA); // Act - Optional key = result.getPartitionKey(); + Throwable thrown = catchThrowable(result::getPartitionKey); // Assert - assertThat(key).isNotPresent(); + assertThat(thrown).isInstanceOf(IllegalStateException.class); } @Test @@ -520,7 +521,7 @@ public void getClusteringKey_RequiredValuesGiven_ShouldReturnClusteringKey() { } @Test - public void getClusteringKey_NotRequiredValuesGiven_ShouldReturnClusteringKey() { + public void getClusteringKey_NotRequiredValuesGiven_ShouldThrowIllegalStateException() { // Arrange ResultImpl result = new ResultImpl( @@ -530,9 +531,9 @@ public void getClusteringKey_NotRequiredValuesGiven_ShouldReturnClusteringKey() TABLE_METADATA); // Act - Optional key = result.getClusteringKey(); + Throwable thrown = catchThrowable(result::getClusteringKey); // Assert - assertThat(key).isNotPresent(); + assertThat(thrown).isInstanceOf(IllegalStateException.class); } } diff --git a/core/src/test/java/com/scalar/db/storage/cassandra/CassandraTest.java b/core/src/test/java/com/scalar/db/storage/cassandra/CassandraTest.java new file mode 100644 index 0000000000..65aef5d4dc --- /dev/null +++ b/core/src/test/java/com/scalar/db/storage/cassandra/CassandraTest.java @@ -0,0 +1,156 @@ +package com.scalar.db.storage.cassandra; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.Session; +import com.scalar.db.api.ConditionBuilder; +import com.scalar.db.api.ConditionalExpression; +import com.scalar.db.api.Scan; +import com.scalar.db.api.Scanner; +import com.scalar.db.api.TableMetadata; +import com.scalar.db.common.FilterableScanner; +import com.scalar.db.common.TableMetadataManager; +import com.scalar.db.common.checker.OperationChecker; +import com.scalar.db.config.DatabaseConfig; +import com.scalar.db.exception.storage.ExecutionException; +import java.util.Properties; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +public class CassandraTest { + + private static final String METADATA_KEYSPACE = "scalardb"; + private static final int ANY_LIMIT = 100; + + private Cassandra cassandra; + @Mock private ClusterManager clusterManager; + @Mock private Session cassandraSession; + @Mock private StatementHandlerManager handlers; + @Mock private SelectStatementHandler handler; + @Mock private TableMetadataManager metadataManager; + @Mock private OperationChecker operationChecker; + @Mock private TableMetadata tableMetadata; + @Mock private ResultSet resultSet; + + @BeforeEach + public void setUp() throws Exception { + MockitoAnnotations.openMocks(this).close(); + when(clusterManager.getSession()).thenReturn(cassandraSession); + Properties cassandraConfigProperties = new Properties(); + cassandraConfigProperties.setProperty(DatabaseConfig.SYSTEM_NAMESPACE_NAME, METADATA_KEYSPACE); + cassandra = + new Cassandra( + new DatabaseConfig(cassandraConfigProperties), + clusterManager, + handlers, + null, + metadataManager, + operationChecker); + } + + @Test + public void scan_WithLimitWithoutConjunction_ShouldHandledWithLimit() throws ExecutionException { + // Arrange + Scan scan = Scan.newBuilder().namespace("ns").table("tbl").all().limit(ANY_LIMIT).build(); + when(handlers.select()).thenReturn(handler); + when(handler.handle(any(Scan.class))).thenReturn(resultSet); + when(metadataManager.getTableMetadata(any(Scan.class))).thenReturn(tableMetadata); + + // Act + Scanner actual = cassandra.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(ScannerImpl.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(handler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getLimit()).isEqualTo(ANY_LIMIT); + } + + @Test + public void scan_WithLimitAndConjunction_ShouldHandledWithoutLimit() throws ExecutionException { + // Arrange + Scan scan = + Scan.newBuilder() + .namespace("ns") + .table("tbl") + .all() + .where(mock(ConditionalExpression.class)) + .limit(ANY_LIMIT) + .build(); + when(handlers.select()).thenReturn(handler); + when(handler.handle(any(Scan.class))).thenReturn(resultSet); + when(metadataManager.getTableMetadata(any(Scan.class))).thenReturn(tableMetadata); + + // Act + Scanner actual = cassandra.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(FilterableScanner.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(handler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getLimit()).isEqualTo(0); + } + + @Test + public void scan_WithConjunctionWithoutProjections_ShouldHandledWithoutProjections() + throws ExecutionException { + // Arrange + Scan scan = + Scan.newBuilder() + .namespace("ns") + .table("tbl") + .all() + .where(ConditionBuilder.column("col2").isLessThanInt(0)) + .build(); + when(handlers.select()).thenReturn(handler); + when(handler.handle(any(Scan.class))).thenReturn(resultSet); + when(metadataManager.getTableMetadata(any(Scan.class))).thenReturn(tableMetadata); + + // Act + Scanner actual = cassandra.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(FilterableScanner.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(handler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getProjections()).isEmpty(); + } + + @Test + public void scan_WithConjunctionAndProjections_ShouldHandledWithExtendedProjections() + throws ExecutionException { + // Arrange + Scan scan = + Scan.newBuilder() + .namespace("ns") + .table("tbl") + .all() + .projections("col1") + .where(ConditionBuilder.column("col2").isLessThanInt(0)) + .build(); + when(handlers.select()).thenReturn(handler); + when(handler.handle(any(Scan.class))).thenReturn(resultSet); + when(metadataManager.getTableMetadata(any(Scan.class))).thenReturn(tableMetadata); + + // Act + Scanner actual = cassandra.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(FilterableScanner.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(handler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getProjections()).containsExactlyInAnyOrder("col1", "col2"); + } +} diff --git a/core/src/test/java/com/scalar/db/storage/cosmos/CosmosTest.java b/core/src/test/java/com/scalar/db/storage/cosmos/CosmosTest.java new file mode 100644 index 0000000000..1e9cf8ab8f --- /dev/null +++ b/core/src/test/java/com/scalar/db/storage/cosmos/CosmosTest.java @@ -0,0 +1,142 @@ +package com.scalar.db.storage.cosmos; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.azure.cosmos.CosmosClient; +import com.scalar.db.api.ConditionBuilder; +import com.scalar.db.api.ConditionalExpression; +import com.scalar.db.api.Scan; +import com.scalar.db.api.Scanner; +import com.scalar.db.common.FilterableScanner; +import com.scalar.db.common.checker.OperationChecker; +import com.scalar.db.config.DatabaseConfig; +import com.scalar.db.exception.storage.ExecutionException; +import com.scalar.db.storage.cassandra.ScannerImpl; +import java.util.Properties; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +public class CosmosTest { + + private static final int ANY_LIMIT = 100; + + private Cosmos cosmos; + @Mock private CosmosClient cosmosClient; + @Mock private SelectStatementHandler selectStatementHandler; + @Mock private PutStatementHandler putStatementHandler; + @Mock private DeleteStatementHandler deleteStatementHandler; + @Mock private BatchHandler batchHandler; + @Mock private OperationChecker operationChecker; + @Mock private ScannerImpl scanner; + + @BeforeEach + public void setUp() throws Exception { + MockitoAnnotations.openMocks(this).close(); + Properties cosmosConfigProperties = new Properties(); + cosmos = + new Cosmos( + new DatabaseConfig(cosmosConfigProperties), + cosmosClient, + selectStatementHandler, + putStatementHandler, + deleteStatementHandler, + batchHandler, + operationChecker); + } + + @Test + public void scan_WithLimitWithoutConjunction_ShouldHandledWithLimit() throws ExecutionException { + // Arrange + Scan scan = Scan.newBuilder().namespace("ns").table("tbl").all().limit(ANY_LIMIT).build(); + when(selectStatementHandler.handle(scan)).thenReturn(scanner); + + // Act + Scanner actual = cosmos.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(ScannerImpl.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(selectStatementHandler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getLimit()).isEqualTo(ANY_LIMIT); + } + + @Test + public void scan_WithLimitAndConjunction_ShouldHandledWithoutLimit() throws ExecutionException { + // Arrange + Scan scan = + Scan.newBuilder() + .namespace("ns") + .table("tbl") + .all() + .where(mock(ConditionalExpression.class)) + .limit(ANY_LIMIT) + .build(); + when(selectStatementHandler.handle(scan)).thenReturn(scanner); + + // Act + Scanner actual = cosmos.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(FilterableScanner.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(selectStatementHandler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getLimit()).isEqualTo(0); + } + + @Test + public void scan_WithConjunctionWithoutProjections_ShouldHandledWithoutProjections() + throws ExecutionException { + // Arrange + Scan scan = + Scan.newBuilder() + .namespace("ns") + .table("tbl") + .all() + .where(ConditionBuilder.column("col2").isLessThanInt(0)) + .build(); + when(selectStatementHandler.handle(scan)).thenReturn(scanner); + + // Act + Scanner actual = cosmos.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(FilterableScanner.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(selectStatementHandler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getProjections()).isEmpty(); + } + + @Test + public void scan_WithConjunctionAndProjections_ShouldHandledWithExtendedProjections() + throws ExecutionException { + // Arrange + Scan scan = + Scan.newBuilder() + .namespace("ns") + .table("tbl") + .all() + .projections("col1") + .where(ConditionBuilder.column("col2").isLessThanInt(0)) + .build(); + when(selectStatementHandler.handle(scan)).thenReturn(scanner); + + // Act + Scanner actual = cosmos.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(FilterableScanner.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(selectStatementHandler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getProjections()).containsExactlyInAnyOrder("col1", "col2"); + } +} diff --git a/core/src/test/java/com/scalar/db/storage/dynamo/DynamoTest.java b/core/src/test/java/com/scalar/db/storage/dynamo/DynamoTest.java new file mode 100644 index 0000000000..400f2b71de --- /dev/null +++ b/core/src/test/java/com/scalar/db/storage/dynamo/DynamoTest.java @@ -0,0 +1,142 @@ +package com.scalar.db.storage.dynamo; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.scalar.db.api.ConditionBuilder; +import com.scalar.db.api.ConditionalExpression; +import com.scalar.db.api.Scan; +import com.scalar.db.api.Scanner; +import com.scalar.db.common.FilterableScanner; +import com.scalar.db.common.checker.OperationChecker; +import com.scalar.db.config.DatabaseConfig; +import com.scalar.db.exception.storage.ExecutionException; +import com.scalar.db.storage.cassandra.ScannerImpl; +import java.util.Properties; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; + +public class DynamoTest { + + private static final int ANY_LIMIT = 100; + + private Dynamo dynamo; + @Mock private DynamoDbClient dynamoDbClient; + @Mock private SelectStatementHandler selectStatementHandler; + @Mock private PutStatementHandler putStatementHandler; + @Mock private DeleteStatementHandler deleteStatementHandler; + @Mock private BatchHandler batchHandler; + @Mock private OperationChecker operationChecker; + @Mock private ScannerImpl scanner; + + @BeforeEach + public void setUp() throws Exception { + MockitoAnnotations.openMocks(this).close(); + Properties cosmosConfigProperties = new Properties(); + dynamo = + new Dynamo( + new DatabaseConfig(cosmosConfigProperties), + dynamoDbClient, + selectStatementHandler, + putStatementHandler, + deleteStatementHandler, + batchHandler, + operationChecker); + } + + @Test + public void scan_WithLimitWithoutConjunction_ShouldHandledWithLimit() throws ExecutionException { + // Arrange + Scan scan = Scan.newBuilder().namespace("ns").table("tbl").all().limit(ANY_LIMIT).build(); + when(selectStatementHandler.handle(scan)).thenReturn(scanner); + + // Act + Scanner actual = dynamo.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(ScannerImpl.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(selectStatementHandler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getLimit()).isEqualTo(ANY_LIMIT); + } + + @Test + public void scan_WithLimitAndConjunction_ShouldHandledWithoutLimit() throws ExecutionException { + // Arrange + Scan scan = + Scan.newBuilder() + .namespace("ns") + .table("tbl") + .all() + .where(mock(ConditionalExpression.class)) + .limit(ANY_LIMIT) + .build(); + when(selectStatementHandler.handle(scan)).thenReturn(scanner); + + // Act + Scanner actual = dynamo.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(FilterableScanner.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(selectStatementHandler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getLimit()).isEqualTo(0); + } + + @Test + public void scan_WithConjunctionWithoutProjections_ShouldHandledWithoutProjections() + throws ExecutionException { + // Arrange + Scan scan = + Scan.newBuilder() + .namespace("ns") + .table("tbl") + .all() + .where(ConditionBuilder.column("col2").isLessThanInt(0)) + .build(); + when(selectStatementHandler.handle(scan)).thenReturn(scanner); + + // Act + Scanner actual = dynamo.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(FilterableScanner.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(selectStatementHandler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getProjections()).isEmpty(); + } + + @Test + public void scan_WithConjunctionAndProjections_ShouldHandledWithExtendedProjections() + throws ExecutionException { + // Arrange + Scan scan = + Scan.newBuilder() + .namespace("ns") + .table("tbl") + .all() + .projections("col1") + .where(ConditionBuilder.column("col2").isLessThanInt(0)) + .build(); + when(selectStatementHandler.handle(scan)).thenReturn(scanner); + + // Act + Scanner actual = dynamo.scan(scan); + + // Assert + assertThat(actual).isInstanceOf(FilterableScanner.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(Scan.class); + verify(selectStatementHandler).handle(captor.capture()); + Scan actualScan = captor.getValue(); + assertThat(actualScan.getProjections()).containsExactlyInAnyOrder("col1", "col2"); + } +} diff --git a/core/src/test/java/com/scalar/db/transaction/consensuscommit/FilteredResultTest.java b/core/src/test/java/com/scalar/db/transaction/consensuscommit/FilteredResultTest.java index 6ea76b0794..11a36d8c86 100644 --- a/core/src/test/java/com/scalar/db/transaction/consensuscommit/FilteredResultTest.java +++ b/core/src/test/java/com/scalar/db/transaction/consensuscommit/FilteredResultTest.java @@ -1,6 +1,7 @@ package com.scalar.db.transaction.consensuscommit; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.catchThrowable; import com.google.common.collect.ImmutableMap; import com.scalar.db.api.Result; @@ -295,7 +296,8 @@ public void withProjections_ShouldFilterOutUnprojectedColumnsAndTransactionMetaC assertThat(filteredResult.getPartitionKey().get().size()).isEqualTo(1); assertThat(filteredResult.getPartitionKey().get().get().get(0)).isEqualTo(ACCOUNT_ID_VALUE); - assertThat(filteredResult.getClusteringKey()).isNotPresent(); + assertThat(catchThrowable(filteredResult::getClusteringKey)) + .isInstanceOf(IllegalStateException.class); assertThat(filteredResult.getValue(ACCOUNT_ID).isPresent()).isTrue(); assertThat(filteredResult.getValue(ACCOUNT_ID).get()).isEqualTo(ACCOUNT_ID_VALUE); @@ -359,7 +361,8 @@ public void withProjectionsAndIncludeMetadataEnabled_ShouldNotIncludeNonProjecte assertThat(filteredResult.getPartitionKey().get().size()).isEqualTo(1); assertThat(filteredResult.getPartitionKey().get().get().get(0)).isEqualTo(ACCOUNT_ID_VALUE); - assertThat(filteredResult.getClusteringKey()).isNotPresent(); + assertThat(catchThrowable(filteredResult::getClusteringKey)) + .isInstanceOf(IllegalStateException.class); assertThat(filteredResult.getValue(ACCOUNT_ID).isPresent()).isTrue(); assertThat(filteredResult.getValue(ACCOUNT_ID).get()).isEqualTo(ACCOUNT_ID_VALUE); @@ -432,7 +435,8 @@ public void withProjectionsAndIncludeMetadataEnabled_ShouldNotIncludeNonProjecte assertThat(filteredResult.getPartitionKey().get().size()).isEqualTo(1); assertThat(filteredResult.getPartitionKey().get().get().get(0)).isEqualTo(ACCOUNT_ID_VALUE); - assertThat(filteredResult.getClusteringKey()).isNotPresent(); + assertThat(catchThrowable(filteredResult::getClusteringKey)) + .isInstanceOf(IllegalStateException.class); assertThat(filteredResult.getValue(ACCOUNT_ID).isPresent()).isTrue(); assertThat(filteredResult.getValue(ACCOUNT_ID).get()).isEqualTo(ACCOUNT_ID_VALUE); @@ -484,7 +488,8 @@ public void withProjectionsAndIncludeMetadataEnabled_ShouldNotIncludeNonProjecte new FilteredResult(result, Collections.singletonList(ACCOUNT_TYPE), TABLE_METADATA, false); // Act Assert - assertThat(filteredResult.getPartitionKey()).isNotPresent(); + assertThat(catchThrowable(filteredResult::getPartitionKey)) + .isInstanceOf(IllegalStateException.class); assertThat(filteredResult.getClusteringKey()).isPresent(); assertThat(filteredResult.getClusteringKey().get().size()).isEqualTo(1); @@ -541,8 +546,10 @@ public void withProjectionsAndIncludeMetadataEnabled_ShouldNotIncludeNonProjecte new FilteredResult(result, Collections.singletonList(BALANCE), TABLE_METADATA, false); // Act Assert - assertThat(filteredResult.getPartitionKey()).isNotPresent(); - assertThat(filteredResult.getClusteringKey()).isNotPresent(); + assertThat(catchThrowable(filteredResult::getPartitionKey)) + .isInstanceOf(IllegalStateException.class); + assertThat(catchThrowable(filteredResult::getClusteringKey)) + .isInstanceOf(IllegalStateException.class); assertThat(filteredResult.getValue(ACCOUNT_ID)).isNotPresent(); assertThat(filteredResult.getValue(ACCOUNT_TYPE)).isNotPresent(); diff --git a/core/src/test/java/com/scalar/db/transaction/consensuscommit/SnapshotTest.java b/core/src/test/java/com/scalar/db/transaction/consensuscommit/SnapshotTest.java index b692445966..ace5c4d859 100644 --- a/core/src/test/java/com/scalar/db/transaction/consensuscommit/SnapshotTest.java +++ b/core/src/test/java/com/scalar/db/transaction/consensuscommit/SnapshotTest.java @@ -13,7 +13,6 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import com.scalar.db.api.ConditionBuilder; @@ -22,7 +21,6 @@ import com.scalar.db.api.Delete; import com.scalar.db.api.DistributedStorage; import com.scalar.db.api.Get; -import com.scalar.db.api.LikeExpression; import com.scalar.db.api.Put; import com.scalar.db.api.Result; import com.scalar.db.api.Scan; @@ -269,22 +267,6 @@ private Delete prepareAnotherDelete() { .forTable(ANY_TABLE_NAME); } - private LikeExpression prepareLike(String pattern) { - return ConditionBuilder.column("col1").isLikeText(pattern); - } - - private LikeExpression prepareLike(String pattern, String escape) { - return ConditionBuilder.column("col1").isLikeText(pattern, escape); - } - - private LikeExpression prepareNotLike(String pattern) { - return ConditionBuilder.column("col1").isNotLikeText(pattern); - } - - private LikeExpression prepareNotLike(String pattern, String escape) { - return ConditionBuilder.column("col1").isNotLikeText(pattern, escape); - } - private void configureBehavior() throws ExecutionException { doNothing().when(prepareComposer).add(any(Put.class), any(TransactionResult.class)); doNothing().when(prepareComposer).add(any(Delete.class), any(TransactionResult.class)); @@ -1675,163 +1657,4 @@ public void verify_CrossPartitionScanGivenAndPutInDifferentTable_ShouldNotThrowE // Assert assertThat(thrown).isInstanceOf(IllegalArgumentException.class); } - - @Test - public void isMatchedWith_SomePatternsWithoutEscapeGiven_ShouldReturnBooleanProperly() { - // Arrange - snapshot = prepareSnapshot(Isolation.SERIALIZABLE); - - // Act Assert - // The following tests are added referring to the similar tests in Spark. - // https://github.com/apache/spark/blob/master/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/RegexpExpressionsSuite.scala - // simple patterns - assertThat(snapshot.isMatched(prepareLike("abdef"), "abdef")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("a\\__b"), "a_%b")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("a_%b"), "addb")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("a\\__b"), "addb")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("a%\\%b"), "addb")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("a%\\%b"), "a_%b")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("a%"), "addb")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("**"), "addb")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("a%"), "abc")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("b%"), "abc")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("bc%"), "abc")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("a_b"), "a\nb")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("a%b"), "ab")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("a%b"), "a\nb")).isTrue(); - - // empty input - assertThat(snapshot.isMatched(prepareLike(""), "")).isTrue(); - assertThat(snapshot.isMatched(prepareLike(""), "a")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("a"), "")).isFalse(); - - // SI-17647 double-escaping backslash - assertThat(snapshot.isMatched(prepareLike("%\\\\%"), "\\\\\\\\")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("%%"), "%%")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("\\\\\\__"), "\\__")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("%\\\\%\\%"), "\\\\\\__")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("%\\\\"), "_\\\\\\%")).isFalse(); - - // unicode - assertThat(snapshot.isMatched(prepareLike("_\u20AC_"), "a\u20ACa")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("_€_"), "a€a")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("_\u20AC_"), "a€a")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("_€_"), "a\u20ACa")).isTrue(); - - // case - assertThat(snapshot.isMatched(prepareLike("a%"), "A")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("A%"), "a")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("_a_"), "AaA")).isTrue(); - - // example - assertThat( - snapshot.isMatched( - prepareLike("\\%SystemDrive\\%\\\\Users%"), "%SystemDrive%\\Users\\John")) - .isTrue(); - } - - @Test - public void isMatchedWith_SomePatternsWithEscapeGiven_ShouldReturnBooleanProperly() { - // Arrange - snapshot = prepareSnapshot(Isolation.SERIALIZABLE); - - // Act Assert - // The following tests are added referring to the similar tests in Spark. - // https://github.com/apache/spark/blob/master/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/RegexpExpressionsSuite.scala - ImmutableList.of("/", "#", "\"") - .forEach( - escape -> { - // simple patterns - assertThat(snapshot.isMatched(prepareLike("abdef", escape), "abdef")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("a" + escape + "__b", escape), "a_%b")) - .isTrue(); - assertThat(snapshot.isMatched(prepareLike("a_%b", escape), "addb")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("a" + escape + "__b", escape), "addb")) - .isFalse(); - assertThat(snapshot.isMatched(prepareLike("a%" + escape + "%b", escape), "addb")) - .isFalse(); - assertThat(snapshot.isMatched(prepareLike("a%" + escape + "%b", escape), "a_%b")) - .isTrue(); - assertThat(snapshot.isMatched(prepareLike("a%", escape), "addb")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("**", escape), "addb")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("a%", escape), "abc")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("b%", escape), "abc")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("bc%", escape), "abc")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("a_b", escape), "a\nb")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("a%b", escape), "ab")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("a%b", escape), "a\nb")).isTrue(); - - // empty input - assertThat(snapshot.isMatched(prepareLike("", escape), "")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("", escape), "a")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("a", escape), "")).isFalse(); - - // SI-17647 double-escaping backslash - assertThat( - snapshot.isMatched( - prepareLike(String.format("%%%s%s%%", escape, escape), escape), - String.format("%s%s%s%s", escape, escape, escape, escape))) - .isTrue(); - assertThat(snapshot.isMatched(prepareLike("%%", escape), "%%")).isTrue(); - assertThat( - snapshot.isMatched( - prepareLike(String.format("%s%s%s__", escape, escape, escape), escape), - String.format("%s__", escape))) - .isTrue(); - assertThat( - snapshot.isMatched( - prepareLike( - String.format("%%%s%s%%%s%%", escape, escape, escape), escape), - String.format("%s%s%s__", escape, escape, escape))) - .isFalse(); - assertThat( - snapshot.isMatched( - prepareLike(String.format("%%%s%s", escape, escape), escape), - String.format("_%s%s%s%%", escape, escape, escape))) - .isFalse(); - - // unicode - assertThat(snapshot.isMatched(prepareLike("_\u20AC_", escape), "a\u20ACa")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("_€_", escape), "a€a")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("_\u20AC_", escape), "a€a")).isTrue(); - assertThat(snapshot.isMatched(prepareLike("_€_", escape), "a\u20ACa")).isTrue(); - - // case - assertThat(snapshot.isMatched(prepareLike("a%", escape), "A")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("A%", escape), "a")).isFalse(); - assertThat(snapshot.isMatched(prepareLike("_a_", escape), "AaA")).isTrue(); - - // example - assertThat( - snapshot.isMatched( - prepareLike( - String.format( - "%s%%SystemDrive%s%%%s%sUsers%%", escape, escape, escape, escape), - escape), - String.format("%%SystemDrive%%%sUsers%sJohn", escape, escape))) - .isTrue(); - }); - } - - @Test - public void isMatchedWith_IsNotLikeOperatorWithSomePatternsGiven_ShouldReturnBooleanProperly() { - // Arrange - snapshot = prepareSnapshot(Isolation.SERIALIZABLE); - - // Act Assert - assertThat(snapshot.isMatched(prepareNotLike("abdef"), "abdef")).isFalse(); - assertThat(snapshot.isMatched(prepareNotLike("a\\__b"), "a_%b")).isFalse(); - assertThat(snapshot.isMatched(prepareNotLike("a_%b"), "addb")).isFalse(); - assertThat(snapshot.isMatched(prepareNotLike("a\\__b"), "addb")).isTrue(); - ImmutableList.of("/", "#", "\"") - .forEach( - escape -> { - assertThat(snapshot.isMatched(prepareNotLike("abdef", escape), "abdef")).isFalse(); - assertThat(snapshot.isMatched(prepareNotLike("a" + escape + "__b", escape), "a_%b")) - .isFalse(); - assertThat(snapshot.isMatched(prepareNotLike("a_%b", escape), "addb")).isFalse(); - assertThat(snapshot.isMatched(prepareNotLike("a" + escape + "__b", escape), "addb")) - .isTrue(); - }); - } } diff --git a/core/src/test/java/com/scalar/db/util/ScalarDbUtilsTest.java b/core/src/test/java/com/scalar/db/util/ScalarDbUtilsTest.java index a5d4751af4..c22645f803 100644 --- a/core/src/test/java/com/scalar/db/util/ScalarDbUtilsTest.java +++ b/core/src/test/java/com/scalar/db/util/ScalarDbUtilsTest.java @@ -4,10 +4,12 @@ import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import com.google.common.collect.ImmutableList; import com.scalar.db.api.ConditionBuilder; import com.scalar.db.api.Delete; import com.scalar.db.api.Get; import com.scalar.db.api.Insert; +import com.scalar.db.api.LikeExpression; import com.scalar.db.api.Mutation; import com.scalar.db.api.Put; import com.scalar.db.api.Scan; @@ -258,4 +260,236 @@ public void checkUpdate_ShouldBehaveProperly() { assertThatThrownBy(() -> ScalarDbUtils.checkUpdate(updateWithInvalidCondition)) .isInstanceOf(IllegalArgumentException.class); } + + @Test + public void isMatchedWith_SomePatternsWithoutEscapeGiven_ShouldReturnBooleanProperly() { + // Arrange Act Assert + // The following tests are added referring to the similar tests in Spark. + // https://github.com/apache/spark/blob/master/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/RegexpExpressionsSuite.scala + // simple patterns + assertThat(ScalarDbUtils.stringMatchesLikeExpression("abdef", prepareLike("abdef"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a_%b", prepareLike("a\\__b"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("addb", prepareLike("a_%b"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("addb", prepareLike("a\\__b"))).isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("addb", prepareLike("a%\\%b"))).isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a_%b", prepareLike("a%\\%b"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("addb", prepareLike("a%"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("addb", prepareLike("**"))).isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("abc", prepareLike("a%"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("abc", prepareLike("b%"))).isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("abc", prepareLike("bc%"))).isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a\nb", prepareLike("a_b"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("ab", prepareLike("a%b"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a\nb", prepareLike("a%b"))).isTrue(); + + // empty input + assertThat(ScalarDbUtils.stringMatchesLikeExpression("", prepareLike(""))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a", prepareLike(""))).isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("", prepareLike("a"))).isFalse(); + + // SI-17647 double-escaping backslash + assertThat(ScalarDbUtils.stringMatchesLikeExpression("\\\\\\\\", prepareLike("%\\\\%"))) + .isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("%%", prepareLike("%%"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("\\__", prepareLike("\\\\\\__"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("\\\\\\__", prepareLike("%\\\\%\\%"))) + .isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("_\\\\\\%", prepareLike("%\\\\"))) + .isFalse(); + + // unicode + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a\u20ACa", prepareLike("_\u20AC_"))) + .isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a€a", prepareLike("_€_"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a€a", prepareLike("_\u20AC_"))).isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a\u20ACa", prepareLike("_€_"))).isTrue(); + + // case + assertThat(ScalarDbUtils.stringMatchesLikeExpression("A", prepareLike("a%"))).isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a", prepareLike("A%"))).isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("AaA", prepareLike("_a_"))).isTrue(); + + // example + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "%SystemDrive%\\Users\\John", prepareLike("\\%SystemDrive\\%\\\\Users%"))) + .isTrue(); + } + + @Test + public void isMatchedWith_SomePatternsWithEscapeGiven_ShouldReturnBooleanProperly() { + // Arrange Act Assert + // The following tests are added referring to the similar tests in Spark. + // https://github.com/apache/spark/blob/master/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/RegexpExpressionsSuite.scala + ImmutableList.of("/", "#", "\"") + .forEach( + escape -> { + // simple patterns + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "abdef", prepareLike("abdef", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "a_%b", prepareLike("a" + escape + "__b", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "addb", prepareLike("a_%b", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "addb", prepareLike("a" + escape + "__b", escape))) + .isFalse(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "addb", prepareLike("a%" + escape + "%b", escape))) + .isFalse(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "a_%b", prepareLike("a%" + escape + "%b", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression("addb", prepareLike("a%", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression("addb", prepareLike("**", escape))) + .isFalse(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression("abc", prepareLike("a%", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression("abc", prepareLike("b%", escape))) + .isFalse(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression("abc", prepareLike("bc%", escape))) + .isFalse(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression("a\nb", prepareLike("a_b", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression("ab", prepareLike("a%b", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression("a\nb", prepareLike("a%b", escape))) + .isTrue(); + + // empty input + assertThat(ScalarDbUtils.stringMatchesLikeExpression("", prepareLike("", escape))) + .isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a", prepareLike("", escape))) + .isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("", prepareLike("a", escape))) + .isFalse(); + + // SI-17647 double-escaping backslash + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + String.format("%s%s%s%s", escape, escape, escape, escape), + prepareLike(String.format("%%%s%s%%", escape, escape), escape))) + .isTrue(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("%%", prepareLike("%%", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + String.format("%s__", escape), + prepareLike(String.format("%s%s%s__", escape, escape, escape), escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + String.format("%s%s%s__", escape, escape, escape), + prepareLike( + String.format("%%%s%s%%%s%%", escape, escape, escape), escape))) + .isFalse(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + String.format("_%s%s%s%%", escape, escape, escape), + prepareLike(String.format("%%%s%s", escape, escape), escape))) + .isFalse(); + + // unicode + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "a\u20ACa", prepareLike("_\u20AC_", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression("a€a", prepareLike("_€_", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "a€a", prepareLike("_\u20AC_", escape))) + .isTrue(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "a\u20ACa", prepareLike("_€_", escape))) + .isTrue(); + + // case + assertThat(ScalarDbUtils.stringMatchesLikeExpression("A", prepareLike("a%", escape))) + .isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a", prepareLike("A%", escape))) + .isFalse(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression("AaA", prepareLike("_a_", escape))) + .isTrue(); + + // example + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + String.format("%%SystemDrive%%%sUsers%sJohn", escape, escape), + prepareLike( + String.format( + "%s%%SystemDrive%s%%%s%sUsers%%", escape, escape, escape, escape), + escape))) + .isTrue(); + }); + } + + @Test + public void isMatchedWith_IsNotLikeOperatorWithSomePatternsGiven_ShouldReturnBooleanProperly() { + // Arrange Act Assert + assertThat(ScalarDbUtils.stringMatchesLikeExpression("abdef", prepareNotLike("abdef"))) + .isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("a_%b", prepareNotLike("a\\__b"))) + .isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("addb", prepareNotLike("a_%b"))).isFalse(); + assertThat(ScalarDbUtils.stringMatchesLikeExpression("addb", prepareNotLike("a\\__b"))) + .isTrue(); + ImmutableList.of("/", "#", "\"") + .forEach( + escape -> { + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "abdef", prepareNotLike("abdef", escape))) + .isFalse(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "a_%b", prepareNotLike("a" + escape + "__b", escape))) + .isFalse(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "addb", prepareNotLike("a_%b", escape))) + .isFalse(); + assertThat( + ScalarDbUtils.stringMatchesLikeExpression( + "addb", prepareNotLike("a" + escape + "__b", escape))) + .isTrue(); + }); + } + + private LikeExpression prepareLike(String pattern) { + return ConditionBuilder.column("col1").isLikeText(pattern); + } + + private LikeExpression prepareLike(String pattern, String escape) { + return ConditionBuilder.column("col1").isLikeText(pattern, escape); + } + + private LikeExpression prepareNotLike(String pattern) { + return ConditionBuilder.column("col1").isNotLikeText(pattern); + } + + private LikeExpression prepareNotLike(String pattern, String escape) { + return ConditionBuilder.column("col1").isNotLikeText(pattern, escape); + } } diff --git a/docs/api-guide.md b/docs/api-guide.md index 38c598bf81..c7aa14dc78 100644 --- a/docs/api-guide.md +++ b/docs/api-guide.md @@ -714,14 +714,14 @@ List results = transaction.scan(scan); {% capture notice--info %} **Note** -You can't specify any filtering conditions and orderings in cross-partition `Scan` except for when using JDBC databases. For details on how to use cross-partition `Scan` with filtering or ordering for JDBC databases, see [Execute cross-partition `Scan` with filtering and ordering](#execute-cross-partition-scan-with-filtering-and-ordering). +You can't specify any orderings in cross-partition `Scan` when using non-JDBC databases. For details on how to use cross-partition `Scan` with filtering or ordering, see [Execute cross-partition `Scan` with filtering and ordering](#execute-cross-partition-scan-with-filtering-and-ordering). {% endcapture %}
{{ notice--info | markdownify }}
##### Execute cross-partition `Scan` with filtering and ordering -By enabling the cross-partition scan option with filtering and ordering for JDBC databases as follows, you can execute a cross-partition `Scan` operation with flexible conditions and orderings: +By enabling the cross-partition scan option with filtering and ordering as follows, you can execute a cross-partition `Scan` operation with flexible conditions and orderings: ```properties scalar.db.cross_partition_scan.enabled=true diff --git a/docs/configurations.md b/docs/configurations.md index 1d457b11b0..7ef8a65168 100644 --- a/docs/configurations.md +++ b/docs/configurations.md @@ -157,7 +157,7 @@ For details about client configurations, see the ScalarDB Cluster [client config ## Cross-partition scan configurations -By enabling the cross-partition scan option below, the `Scan` operation can retrieve all records across partitions. In addition, you can specify arbitrary conditions and orderings in the cross-partition `Scan` operation by enabling `cross_partition_scan.filtering` and `cross_partition_scan.ordering`, respectively. Currently, the cross-partition scan with filtering and ordering is available only for JDBC databases. To enable filtering and ordering, `scalar.db.cross_partition_scan.enabled` must be set to `true`. +By enabling the cross-partition scan option as described below, the `Scan` operation can retrieve all records across partitions. In addition, you can specify arbitrary conditions and orderings in the cross-partition `Scan` operation by enabling `cross_partition_scan.filtering` and `cross_partition_scan.ordering`, respectively. Currently, the cross-partition scan with ordering is available only for JDBC databases. To enable filtering and ordering, `scalar.db.cross_partition_scan.enabled` must be set to `true`. For details on how to use cross-partition scan, see [Scan operation](./api-guide.md#scan-operation). diff --git a/integration-test/src/main/java/com/scalar/db/api/DistributedStorageCrossPartitionScanIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/api/DistributedStorageCrossPartitionScanIntegrationTestBase.java index adc3dd027d..4f87fd58d4 100644 --- a/integration-test/src/main/java/com/scalar/db/api/DistributedStorageCrossPartitionScanIntegrationTestBase.java +++ b/integration-test/src/main/java/com/scalar/db/api/DistributedStorageCrossPartitionScanIntegrationTestBase.java @@ -565,7 +565,6 @@ private Scan prepareScanWithLike(boolean isLike, String pattern) { .table(CONDITION_TEST_TABLE) .all() .where(condition) - .ordering(Ordering.asc(PARTITION_KEY_NAME)) .build(); } @@ -579,7 +578,6 @@ private Scan prepareScanWithLike(boolean isLike, String pattern, String escape) .table(CONDITION_TEST_TABLE) .all() .where(condition) - .ordering(Ordering.asc(PARTITION_KEY_NAME)) .build(); } @@ -589,7 +587,7 @@ private void assertScanResult( for (Result actualResult : actualResults) { actual.add(actualResult.getInt(PARTITION_KEY_NAME)); } - assertThat(actual).describedAs(description).isEqualTo(expected); + assertThat(actual).describedAs(description).containsExactlyInAnyOrderElementsOf(expected); } private void assertOrderedScanResult( @@ -605,6 +603,30 @@ private void assertOrderedScanResult( assertThat(actual).describedAs(description).isEqualTo(expected); } + private void assertLimitedScanResult( + List actualResults, List expected, int limit, String description) { + List actual = new ArrayList<>(); + for (Result actualResult : actualResults) { + actual.add(actualResult.getInt(PARTITION_KEY_NAME)); + } + assertThat(actual.size()).describedAs(description).isLessThanOrEqualTo(limit); + assertThat(actual).describedAs(description).containsAnyElementsOf(expected); + } + + private void assertProjectedResult( + List actualResults, + List expected, + List projections, + String description) { + List actual = new ArrayList<>(); + for (Result actualResult : actualResults) { + assertThat(actualResult.getContainedColumnNames()) + .containsExactlyInAnyOrderElementsOf(projections); + actual.add(actualResult.getInt(PARTITION_KEY_NAME)); + } + assertThat(actual).describedAs(description).containsExactlyInAnyOrderElementsOf(expected); + } + private String description(Column column, Operator operator) { return String.format("failed with column: %s, operator: %s", column, operator); } @@ -613,6 +635,10 @@ private String description(Column column, Operator operator, int value) { return description(column, operator) + String.format(", value: %s", value); } + private String descriptionForLimitTests(Column column, Operator operator, int limit) { + return description(column, operator) + String.format(", limit: %s", limit); + } + private String description( DataType firstColumnType, Order firstColumnOrder, @@ -762,7 +788,6 @@ private void scan_WithCondition_ShouldReturnProperResult( .table(CONDITION_TEST_TABLE) .all() .where(ConditionBuilder.buildConditionalExpression(column, operator)) - .ordering(Ordering.asc(PARTITION_KEY_NAME)) .build(); // Act @@ -784,7 +809,6 @@ private void scan_WithNullCondition_ShouldReturnProperResult(Column column, O .table(CONDITION_TEST_TABLE) .all() .where(ConditionBuilder.buildConditionalExpression(column, operator)) - .ordering(Ordering.asc(PARTITION_KEY_NAME)) .build(); // Act @@ -816,14 +840,15 @@ public void scan_WithConjunctiveNormalFormConditionsShouldReturnProperResult() .map(i -> prepareOrConditionSet(ImmutableList.of(columns1.get(i), columns2.get(i)))) .collect(Collectors.toList()); orConditionSets.forEach(builder::and); - builder.ordering(Ordering.asc(PARTITION_KEY_NAME)); // Act List actual = scanAll(builder.build()); // Assert assertScanResult( - actual, getExpectedResults(DataType.INT, Operator.LTE, 2), "failed with CNF conditions"); + actual, + getExpectedResults(DataType.INT, Operator.LTE, CONDITION_TEST_PREDICATE_VALUE), + "failed with CNF conditions"); } @Test @@ -838,7 +863,6 @@ public void scan_WithDisjunctiveNormalFormConditionsShouldReturnProperResult() .all() .where(prepareAndConditionSet(prepareNonKeyColumns(1))) .or(prepareAndConditionSet(prepareNonKeyColumns(2))) - .ordering(Ordering.asc(PARTITION_KEY_NAME)) .build(); // Act @@ -846,7 +870,9 @@ public void scan_WithDisjunctiveNormalFormConditionsShouldReturnProperResult() // Assert assertScanResult( - actual, getExpectedResults(DataType.INT, Operator.LTE, 2), "failed with DNF conditions"); + actual, + getExpectedResults(DataType.INT, Operator.LTE, CONDITION_TEST_PREDICATE_VALUE), + "failed with DNF conditions"); } @Test @@ -1031,6 +1057,74 @@ private void scan_WithLikeCondition_ShouldReturnProperResult( assertScanResult(actual, expected, description); } + @Test + public void scan_WithConditionAndLimit_ShouldReturnProperResult() + throws java.util.concurrent.ExecutionException, InterruptedException { + prepareRecords(); + + List> testCallables = new ArrayList<>(); + IntStream.range(1, CONDITION_TEST_TABLE_NUM_ROWS + 1) + .forEach( + limit -> { + testCallables.add( + () -> { + scan_WithConditionAndLimit_ShouldReturnProperResult(limit); + return null; + }); + }); + + executeInParallel(testCallables); + } + + private void scan_WithConditionAndLimit_ShouldReturnProperResult(int limit) + throws IOException, ExecutionException { + // Arrange + IntColumn column = IntColumn.of(COL_NAME1, CONDITION_TEST_PREDICATE_VALUE); + Scan scan = + Scan.newBuilder() + .namespace(getNamespaceName()) + .table(CONDITION_TEST_TABLE) + .all() + .where(ConditionBuilder.buildConditionalExpression(column, Operator.LTE)) + .limit(limit) + .build(); + + // Act + List actual = scanAll(scan); + + // Assert + assertLimitedScanResult( + actual, + // expected full (not-limited) results + getExpectedResults(column.getDataType(), Operator.LTE, column.getIntValue()), + limit, + descriptionForLimitTests(column, Operator.LTE, limit)); + } + + @Test + public void scan_WithConditionButColumnsNotAppearedInProjections_ShouldReturnProjectedResult() + throws IOException, ExecutionException { + prepareRecords(); + // Arrange + IntColumn column = IntColumn.of(COL_NAME1, CONDITION_TEST_PREDICATE_VALUE); + Scan scan = + Scan.newBuilder() + .namespace(getNamespaceName()) + .table(CONDITION_TEST_TABLE) + .all() + .projections(PARTITION_KEY_NAME, COL_NAME2) + .where(ConditionBuilder.buildConditionalExpression(column, Operator.LTE)) + .build(); + + // Act + List actual = scanAll(scan); + assertProjectedResult( + actual, + getExpectedResults(column.getDataType(), Operator.LTE, column.getIntValue()), + ImmutableList.of(PARTITION_KEY_NAME, COL_NAME2), + description(column, Operator.LTE)); + } + private void executeDdls(List> ddls) throws InterruptedException, java.util.concurrent.ExecutionException { if (isParallelDdlSupported()) { diff --git a/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionCrossPartitionScanIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionCrossPartitionScanIntegrationTestBase.java index 042d8bffb0..0785417605 100644 --- a/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionCrossPartitionScanIntegrationTestBase.java +++ b/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionCrossPartitionScanIntegrationTestBase.java @@ -226,11 +226,7 @@ public void scan_CrossPartitionScanWithLimitGivenForCommittedRecord_ShouldReturn // Arrange populateRecords(); DistributedTransaction transaction = manager.start(); - Scan scan = - Scan.newBuilder(prepareCrossPartitionScan(1, 0, 2)) - .ordering(Ordering.asc(ACCOUNT_TYPE)) - .limit(2) - .build(); + Scan scan = Scan.newBuilder(prepareCrossPartitionScan(1, 0, 2)).limit(2).build(); // Act List results = transaction.scan(scan); @@ -238,15 +234,7 @@ public void scan_CrossPartitionScanWithLimitGivenForCommittedRecord_ShouldReturn // Assert assertThat(results.size()).isEqualTo(2); - assertThat(results.get(0).getInt(ACCOUNT_ID)).isEqualTo(10); - assertThat(results.get(0).getInt(ACCOUNT_TYPE)).isEqualTo(0); - assertThat(getBalance(results.get(0))).isEqualTo(INITIAL_BALANCE); - assertThat(results.get(0).getInt(SOME_COLUMN)).isEqualTo(0); - - assertThat(results.get(1).getInt(ACCOUNT_ID)).isEqualTo(11); - assertThat(results.get(1).getInt(ACCOUNT_TYPE)).isEqualTo(1); - assertThat(getBalance(results.get(1))).isEqualTo(INITIAL_BALANCE); - assertThat(results.get(1).getInt(SOME_COLUMN)).isEqualTo(1); + TestUtils.assertResultsAreASubsetOf(results, prepareExpectedResults(1, 0, 2, true)); } @Test @@ -430,7 +418,6 @@ protected Scan prepareCrossPartitionScanWithLike(boolean isLike, String pattern) .table(TABLE_WITH_TEXT) .all() .where(condition) - .ordering(Ordering.asc(ACCOUNT_ID)) .consistency(Consistency.LINEARIZABLE) .build(); } @@ -445,7 +432,6 @@ protected Scan prepareCrossPartitionScanWithLike(boolean isLike, String pattern, .table(TABLE_WITH_TEXT) .all() .where(condition) - .ordering(Ordering.asc(ACCOUNT_ID)) .consistency(Consistency.LINEARIZABLE) .build(); } @@ -480,6 +466,6 @@ private void assertScanResult(List actualResults, List expected for (Result actualResult : actualResults) { actual.add(actualResult.getInt(ACCOUNT_ID)); } - assertThat(actual).isEqualTo(expected); + assertThat(actual).containsExactlyInAnyOrderElementsOf(expected); } } diff --git a/integration-test/src/main/java/com/scalar/db/api/TwoPhaseCommitTransactionCrossPartitionScanIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/api/TwoPhaseCommitTransactionCrossPartitionScanIntegrationTestBase.java index 066996b269..0ed5740fc2 100644 --- a/integration-test/src/main/java/com/scalar/db/api/TwoPhaseCommitTransactionCrossPartitionScanIntegrationTestBase.java +++ b/integration-test/src/main/java/com/scalar/db/api/TwoPhaseCommitTransactionCrossPartitionScanIntegrationTestBase.java @@ -403,7 +403,6 @@ protected Scan prepareCrossPartitionScanWithLike( .table(tableName) .all() .where(condition) - .ordering(Ordering.asc(ACCOUNT_ID)) .consistency(Consistency.LINEARIZABLE) .build(); } @@ -419,7 +418,6 @@ protected Scan prepareCrossPartitionScanWithLike( .table(tableName) .all() .where(condition) - .ordering(Ordering.asc(ACCOUNT_ID)) .consistency(Consistency.LINEARIZABLE) .build(); } @@ -454,6 +452,6 @@ private void assertScanResult(List actualResults, List expected for (Result actualResult : actualResults) { actual.add(actualResult.getInt(ACCOUNT_ID)); } - assertThat(actual).isEqualTo(expected); + assertThat(actual).containsExactlyInAnyOrderElementsOf(expected); } }