diff --git a/src/java/org/apache/cassandra/db/virtual/DataPlacementsTable.java b/src/java/org/apache/cassandra/db/virtual/DataPlacementsTable.java new file mode 100644 index 000000000000..8fe3a7ec98ff --- /dev/null +++ b/src/java/org/apache/cassandra/db/virtual/DataPlacementsTable.java @@ -0,0 +1,212 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db.virtual; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; + +import com.google.common.base.Preconditions; + +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.db.marshal.CompositeType; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.SetType; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.dht.LocalPartitioner; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.ReversedLongLocalPartitioner; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.locator.EndpointsForRange; +import org.apache.cassandra.locator.Replica; +import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.schema.ReplicationParams; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.ownership.DataPlacement; +import org.apache.cassandra.tcm.ownership.DataPlacements; +import org.apache.cassandra.tcm.ownership.VersionedEndpoints; + +final class DataPlacementsTable extends AbstractVirtualTable +{ + static final String TABLE_COMMENT = "Data placement information showing read and write endpoints per range"; + + static final String TABLE_NAME = "data_placements"; + + // Partition key columns + static final String KEYSPACE_NAME = "keyspace_name"; + static final String TABLE_NAME_COLUMN = "table_name"; + static final String RANGE_START = "range_start"; + static final String RANGE_END = "range_end"; + + // Regular columns + static final String RANGE_START_BYTES = "range_start_bytes"; + static final String RANGE_END_BYTES = "range_end_bytes"; + static final String TOKEN_TYPE = "token_type"; + static final String READ_ENDPOINTS = "read_endpoints"; + static final String WRITE_ENDPOINTS = "write_endpoints"; + static final String READ_REPLICAS = "read_replicas"; + static final String WRITE_REPLICAS = "write_replicas"; + + DataPlacementsTable(String keyspace) + { + super(buildTableMetadata(keyspace)); + } + + @Override + public DataSet data() + { + SimpleDataSet result = new SimpleDataSet(metadata()); + + ClusterMetadata metadata = ClusterMetadata.current(); + DataPlacements placements = metadata.placements; + + for (KeyspaceMetadata ksm : metadata.schema.getKeyspaces()) + { + ReplicationParams params = ksm.params.replication; + DataPlacement placement = placements.get(params); + + for (TableMetadata table : ksm.tables) + addPlacementRows(result, ksm.name, table.name, params, placement, metadata, null, null); + } + + return result; + } + + @Override + public DataSet data(DecoratedKey partitionKey) + { + SimpleDataSet result = new SimpleDataSet(metadata()); + + ByteBuffer keyBytes = partitionKey.getKey(); + + Preconditions.checkArgument(metadata().partitionKeyType instanceof CompositeType, + "Expected CompositeType partition key, got %s", metadata().partitionKeyType.getClass()); + + ByteBuffer[] components = ((CompositeType) metadata().partitionKeyType).split(keyBytes); + Preconditions.checkArgument(components.length == 1 || components.length == 2 || components.length == 4, + "Expected 1, 2, or 4 partition key components (keyspace[, table[, range_start, range_end]]), got %d", + components.length); + + String keyspaceName = UTF8Type.instance.compose(components[0]); + String tableName = components.length >= 2 ? UTF8Type.instance.compose(components[1]) : null; + String rangeStartFilter = components.length == 4 ? UTF8Type.instance.compose(components[2]) : null; + String rangeEndFilter = components.length == 4 ? UTF8Type.instance.compose(components[3]) : null; + + ClusterMetadata metadata = ClusterMetadata.current(); + KeyspaceMetadata ksm = metadata.schema.getKeyspaces().getNullable(keyspaceName); + if (ksm == null) + return result; + + ReplicationParams params = ksm.params.replication; + DataPlacement placement = metadata.placements.get(params); + Preconditions.checkState(placement != null, "Placements should never be null for %s", keyspaceName); + + if (tableName != null) + { + TableMetadata table = ksm.getTableOrViewNullable(tableName); + Preconditions.checkArgument(table != null, "Couldn't find table %s", tableName); + addPlacementRows(result, ksm.name, table.name, params, placement, metadata, rangeStartFilter, rangeEndFilter); + } + else + { + for (TableMetadata table : ksm.tables) + addPlacementRows(result, ksm.name, table.name, params, placement, metadata, null, null); + } + + return result; + } + + private void addPlacementRows(SimpleDataSet result, String keyspaceName, String tableName, ReplicationParams params, DataPlacement placement, ClusterMetadata metadata, String rangeStartFilter, String rangeEndFilter) + { + Preconditions.checkState(placement.reads.ranges.equals(placement.writes.ranges), + "Read ranges (%s) are not the same as write ranges (%s)", + placement.reads.ranges, placement.writes.ranges); + + List> ranges = placement.reads.ranges; + for (int i = 0; i < ranges.size(); i++) + { + Range range = ranges.get(i); + if (rangeStartFilter != null && !range.left.toString().equals(rangeStartFilter)) + continue; + if (rangeEndFilter != null && !range.right.toString().equals(rangeEndFilter)) + continue; + + VersionedEndpoints.ForRange readEndpoints = placement.reads.forRange(range); + VersionedEndpoints.ForRange writeEndpoints = placement.writes.forRange(range); + + Set readEndpointSet = toEndpointStrings(readEndpoints.get()); + Set writeEndpointSet = toEndpointStrings(writeEndpoints.get()); + + Set readReplicaSet = toNodeIds(readEndpoints.get(), metadata); + Set writeReplicaSet = toNodeIds(writeEndpoints.get(), metadata); + + IPartitioner partitioner = params.isMeta() ? ReversedLongLocalPartitioner.instance : metadata.partitioner; + result.row(keyspaceName, tableName, range.left.toString(), range.right.toString()) + .column(TOKEN_TYPE, range.left.getClass().getCanonicalName()) + .column(RANGE_START_BYTES, partitioner.getTokenFactory().toByteArray(range.left)) + .column(RANGE_END_BYTES, partitioner.getTokenFactory().toByteArray(range.right)) + .column(READ_ENDPOINTS, readEndpointSet) + .column(WRITE_ENDPOINTS, writeEndpointSet) + .column(READ_REPLICAS, readReplicaSet) + .column(WRITE_REPLICAS, writeReplicaSet); + } + } + + static Set toEndpointStrings(EndpointsForRange endpoints) + { + return endpoints.stream() + .map(Replica::endpoint) + .map(Object::toString) + .collect(Collectors.toSet()); + } + + static Set toNodeIds(EndpointsForRange endpoints, ClusterMetadata metadata) + { + return endpoints.stream() + .map(Replica::endpoint) + .map(metadata.directory::peerId) + .map(NodeId::id) + .collect(Collectors.toSet()); + } + + private static TableMetadata buildTableMetadata(String keyspace) + { + return TableMetadata.builder(keyspace, TABLE_NAME) + .comment(TABLE_COMMENT) + .kind(TableMetadata.Kind.VIRTUAL) + .partitioner(new LocalPartitioner(CompositeType.getInstance(UTF8Type.instance, UTF8Type.instance, UTF8Type.instance, UTF8Type.instance))) + .addPartitionKeyColumn(KEYSPACE_NAME, UTF8Type.instance) + .addPartitionKeyColumn(TABLE_NAME_COLUMN, UTF8Type.instance) + .addPartitionKeyColumn(RANGE_START, UTF8Type.instance) + .addPartitionKeyColumn(RANGE_END, UTF8Type.instance) + .addRegularColumn(TOKEN_TYPE, UTF8Type.instance) + .addRegularColumn(RANGE_START_BYTES, BytesType.instance) + .addRegularColumn(RANGE_END_BYTES, BytesType.instance) + .addRegularColumn(READ_ENDPOINTS, SetType.getInstance(UTF8Type.instance, false)) + .addRegularColumn(WRITE_ENDPOINTS, SetType.getInstance(UTF8Type.instance, false)) + .addRegularColumn(READ_REPLICAS, SetType.getInstance(Int32Type.instance, false)) + .addRegularColumn(WRITE_REPLICAS, SetType.getInstance(Int32Type.instance, false)) + .build(); + } +} diff --git a/src/java/org/apache/cassandra/db/virtual/PartitionLocationTable.java b/src/java/org/apache/cassandra/db/virtual/PartitionLocationTable.java new file mode 100644 index 000000000000..af4d239b8828 --- /dev/null +++ b/src/java/org/apache/cassandra/db/virtual/PartitionLocationTable.java @@ -0,0 +1,163 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db.virtual; + +import java.nio.ByteBuffer; +import java.util.Set; + +import com.google.common.base.Preconditions; + +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.db.marshal.CompositeType; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.SetType; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.dht.LocalPartitioner; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.ReversedLongLocalPartitioner; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.schema.ReplicationParams; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ownership.DataPlacement; +import org.apache.cassandra.tcm.ownership.VersionedEndpoints; + +import static com.google.common.base.Preconditions.*; + +final class PartitionLocationTable extends AbstractVirtualTable +{ + static final String TABLE_NAME = "partition_location"; + static final String TABLE_COMMENT = "shows the token range and replicas (read and write) for a given partition"; + + // Partition keys + static final String COLUMN_KEYSPACE_NAME = "keyspace_name"; + static final String COLUMN_TABLE_NAME = "table_name"; + static final String COLUMN_KEY = "key"; + + // Regular columns + static final String COLUMN_TOKEN = "tkn"; // can't use "token" because this is reserved in CQL + static final String COLUMN_TOKEN_BYTES = "tkn_bytes"; + static final String COLUMN_RANGE_START = "range_start"; + static final String COLUMN_RANGE_END = "range_end"; + static final String COLUMN_RANGE_START_BYTES = "range_start_bytes"; + static final String COLUMN_RANGE_END_BYTES = "range_end_bytes"; + static final String COLUMN_READ_ENDPOINTS = "read_endpoints"; + static final String COLUMN_WRITE_ENDPOINTS = "write_endpoints"; + static final String COLUMN_READ_REPLICAS = "read_replicas"; + static final String COLUMN_WRITE_REPLICAS = "write_replicas"; + + PartitionLocationTable(String keyspace) + { + super(buildTableMetadata(keyspace)); + } + + @Override + public DataSet data() + { + throw new InvalidRequestException("Partition location table requires a partition key, for example: " + + "SELECT * FROM system_views.partition_location WHERE keyspace_name = 'ks' AND table_name = 'tbl' AND key = '1:a'"); + } + + @Override + public DataSet data(DecoratedKey partitionKey) + { + SimpleDataSet result = new SimpleDataSet(metadata()); + + // Partition key is (keyspace_name, table_name, key) + ByteBuffer keyBytes = partitionKey.getKey(); + + checkArgument(metadata().partitionKeyType instanceof CompositeType, + "PartitionLocationTable partition key type must be CompositeType, got %s", + metadata().partitionKeyType.getClass().getName()); + + ByteBuffer[] components = ((CompositeType) metadata().partitionKeyType).split(keyBytes); + checkArgument(components.length == 3, + "PartitionLocationTable partition key must have exactly 3 components: keyspace_name, table_name, key; got %d", + components.length); + + String keyspaceName = UTF8Type.instance.compose(components[0]); + String tableName = UTF8Type.instance.compose(components[1]); + String keyString = UTF8Type.instance.compose(components[2]); + + ClusterMetadata metadata = ClusterMetadata.current(); + KeyspaceMetadata ksm = checkNotNull(metadata.schema.getKeyspaceMetadata(keyspaceName), + "Keyspace %s is not found in metadata", keyspaceName); + TableMetadata table = checkNotNull(ksm.getTableOrViewNullable(tableName), + "Table %s is not found in metadata (within keyspace %s)", tableName, keyspaceName); + + ByteBuffer partitionKeyBytes = table.partitionKeyType.fromString(keyString); + + DecoratedKey dk = table.partitioner.decorateKey(partitionKeyBytes); + Token token = dk.getToken(); + + ReplicationParams replicationParams = ksm.params.replication; + DataPlacement placement = metadata.placements.get(replicationParams); + + VersionedEndpoints.ForRange readEndpoints = placement.reads.forRange(token); + VersionedEndpoints.ForRange writeEndpoints = placement.writes.forRange(token); + + Range range = readEndpoints.get().range(); + + Set readEndpointSet = DataPlacementsTable.toEndpointStrings(readEndpoints.get()); + Set writeEndpointSet = DataPlacementsTable.toEndpointStrings(writeEndpoints.get()); + + Set readReplicaSet = DataPlacementsTable.toNodeIds(readEndpoints.get(), metadata); + Set writeReplicaSet = DataPlacementsTable.toNodeIds(writeEndpoints.get(), metadata); + + IPartitioner partitioner = replicationParams.isMeta() ? ReversedLongLocalPartitioner.instance : metadata.partitioner; + result.row(keyspaceName, tableName, keyString) + .column(COLUMN_TOKEN, token.toString()) + .column(COLUMN_TOKEN_BYTES, partitioner.getTokenFactory().toByteArray(token)) + .column(COLUMN_RANGE_START, range.left.toString()) + .column(COLUMN_RANGE_END, range.right.toString()) + .column(COLUMN_RANGE_START_BYTES, partitioner.getTokenFactory().toByteArray(range.left)) + .column(COLUMN_RANGE_END_BYTES, partitioner.getTokenFactory().toByteArray(range.right)) + .column(COLUMN_READ_ENDPOINTS, readEndpointSet) + .column(COLUMN_WRITE_ENDPOINTS, writeEndpointSet) + .column(COLUMN_READ_REPLICAS, readReplicaSet) + .column(COLUMN_WRITE_REPLICAS, writeReplicaSet); + + return result; + } + + private static TableMetadata buildTableMetadata(String keyspace) { + return TableMetadata.builder(keyspace, TABLE_NAME) + .comment(TABLE_COMMENT) + .kind(TableMetadata.Kind.VIRTUAL) + .partitioner(new LocalPartitioner(CompositeType.getInstance(UTF8Type.instance, UTF8Type.instance, UTF8Type.instance))) + .addPartitionKeyColumn(COLUMN_KEYSPACE_NAME, UTF8Type.instance) + .addPartitionKeyColumn(COLUMN_TABLE_NAME, UTF8Type.instance) + .addPartitionKeyColumn(COLUMN_KEY, UTF8Type.instance) + .addRegularColumn(COLUMN_TOKEN, UTF8Type.instance) + .addRegularColumn(COLUMN_TOKEN_BYTES, BytesType.instance) + .addRegularColumn(COLUMN_RANGE_START, UTF8Type.instance) + .addRegularColumn(COLUMN_RANGE_END, UTF8Type.instance) + .addRegularColumn(COLUMN_RANGE_START_BYTES, BytesType.instance) + .addRegularColumn(COLUMN_RANGE_END_BYTES, BytesType.instance) + .addRegularColumn(COLUMN_READ_ENDPOINTS, SetType.getInstance(UTF8Type.instance, false)) + .addRegularColumn(COLUMN_WRITE_ENDPOINTS, SetType.getInstance(UTF8Type.instance, false)) + .addRegularColumn(COLUMN_READ_REPLICAS, SetType.getInstance(Int32Type.instance, false)) + .addRegularColumn(COLUMN_WRITE_REPLICAS, SetType.getInstance(Int32Type.instance, false)) + .build(); + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java index 9a80d1e9b930..d953310165d7 100644 --- a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java @@ -69,6 +69,8 @@ private SystemViewsKeyspace() .add(new LocalTable(VIRTUAL_VIEWS)) .add(new ClusterMetadataLogTable(VIRTUAL_VIEWS)) .add(new ClusterMetadataDirectoryTable(VIRTUAL_VIEWS)) + .add(new DataPlacementsTable(VIRTUAL_VIEWS)) + .add(new PartitionLocationTable(VIRTUAL_VIEWS)) .addAll(LocalRepairTables.getAll(VIRTUAL_VIEWS)) .addAll(CIDRFilteringMetricsTable.getAll(VIRTUAL_VIEWS)) .addAll(StorageAttachedIndexTables.getAll(VIRTUAL_VIEWS)) diff --git a/test/distributed/org/apache/cassandra/distributed/test/DataPlacementsTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/DataPlacementsTableTest.java new file mode 100644 index 000000000000..304ffe358870 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/DataPlacementsTableTest.java @@ -0,0 +1,376 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.locator.Replica; +import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ClusterMetadataService; +import org.apache.cassandra.tcm.membership.Location; +import org.apache.cassandra.tcm.membership.NodeAddresses; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.membership.NodeVersion; +import org.apache.cassandra.tcm.ownership.DataPlacement; +import org.apache.cassandra.tcm.ownership.VersionedEndpoints; +import org.apache.cassandra.tcm.transformations.Register; +import org.apache.cassandra.tcm.transformations.UnsafeJoin; +import org.apache.cassandra.utils.Shared; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +public class DataPlacementsTableTest extends TestBaseImpl +{ + private static final int NUM_FAKE_NODES = 60; + private static final String DC1 = "datacenter0"; + private static final String DC2 = "datacenter1"; + private static final String SIMPLE_KS = "test_ks_simple"; + private static final String NTS_KS = "test_ks_nts"; + + private static final String SIMPLE_TABLE1 = "tbl1"; + private static final String SIMPLE_TABLE2 = "tbl2"; + private static final String NTS_TABLE1 = "nts_tbl1"; + private static final String NTS_TABLE2 = "nts_tbl2"; + + private static final long REAL_NODE_TOKEN = Long.MAX_VALUE - 1; + + private static Cluster CLUSTER; + private static Map> SIMPLE_EXPECTED; + private static Map> NTS_EXPECTED; + + @BeforeClass + public static void setup() throws IOException + { + CLUSTER = Cluster.build(1) + .withConfig(c -> c.set("num_tokens", 1) + .set("initial_token", Long.toString(REAL_NODE_TOKEN))) + .start(); + + // Register 60 fake nodes split evenly across the two DCs, evenly spaced around the ring. + CLUSTER.get(1).runOnInstance(() -> { + try + { + for (int i = 0; i < NUM_FAKE_NODES; i++) + { + String dc = (i % 2 == 0) ? DC1 : DC2; + String address = "127.0.1." + (i + 2); + + InetAddressAndPort addr = InetAddressAndPort.getByName(address); + NodeAddresses nodeAddresses = new NodeAddresses(addr); + Location location = new Location(dc, "rack1"); + + ClusterMetadata metadata = ClusterMetadataService.instance().commit( + new Register(nodeAddresses, location, NodeVersion.CURRENT)); + + NodeId nodeId = metadata.directory.peerId(addr); + + long token = Long.MIN_VALUE + (long) i * (Long.MAX_VALUE / (NUM_FAKE_NODES / 2)); + Set tokens = new HashSet<>(); + tokens.add(new Murmur3Partitioner.LongToken(token)); + + UnsafeJoin.unsafeJoin(nodeId, tokens); + } + } + catch (Exception e) + { + throw new RuntimeException("Failed to register fake nodes", e); + } + }); + + CLUSTER.schemaChange(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3}", SIMPLE_KS)); + CLUSTER.schemaChange(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'NetworkTopologyStrategy', '%s': 2, '%s': 2}", NTS_KS, DC1, DC2)); + + CLUSTER.schemaChange(String.format("CREATE TABLE %s.%s (id int PRIMARY KEY, val text)", SIMPLE_KS, SIMPLE_TABLE1)); + CLUSTER.schemaChange(String.format("CREATE TABLE %s.%s (id int PRIMARY KEY, val text)", SIMPLE_KS, SIMPLE_TABLE2)); + CLUSTER.schemaChange(String.format("CREATE TABLE %s.%s (id int PRIMARY KEY, val text)", NTS_KS, NTS_TABLE1)); + CLUSTER.schemaChange(String.format("CREATE TABLE %s.%s (id int PRIMARY KEY, val text)", NTS_KS, NTS_TABLE2)); + + SIMPLE_EXPECTED = fetchExpectedPlacements(SIMPLE_KS); + NTS_EXPECTED = fetchExpectedPlacements(NTS_KS); + } + + + @AfterClass + public static void cleanup() + { + if (CLUSTER != null) + CLUSTER.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void testPlacements() + { + Map>> allExpected = new HashMap<>(); + allExpected.put(SIMPLE_KS, SIMPLE_EXPECTED); + allExpected.put(NTS_KS, NTS_EXPECTED); + + Object[][] result = CLUSTER.coordinator(1) + .execute("SELECT keyspace_name, table_name, range_start, range_end, token_type, " + + "range_start_bytes, range_end_bytes, read_endpoints, write_endpoints, " + + "read_replicas, write_replicas " + + "FROM system_views.data_placements", + ConsistencyLevel.ONE); + + assertThat(result).isNotEmpty(); + + Map> tablesFound = new HashMap<>(); + Map> rangesFound = new HashMap<>(); + for (String ks : allExpected.keySet()) + { + tablesFound.put(ks, new HashSet<>()); + rangesFound.put(ks, new HashSet<>()); + } + + for (Object[] row : result) + { + String keyspace = (String) row[0]; + Map> expected = allExpected.get(keyspace); + if (expected == null) + continue; // skip system keyspaces + + String tableName = (String) row[1]; + String rangeStart = (String) row[2]; + String rangeEnd = (String) row[3]; + String tokenType = (String) row[4]; + ByteBuffer rangeStartBytes = (ByteBuffer) row[5]; + ByteBuffer rangeEndBytes = (ByteBuffer) row[6]; + Set readEndpoints = (Set) row[7]; + Set writeEndpoints = (Set) row[8]; + Set readReplicas = (Set) row[9]; + Set writeReplicas = (Set) row[10]; + + assertThat(tokenType).as("token_type for %s.%s", keyspace, tableName) + .contains("Murmur3Partitioner"); + assertThat(rangeStartBytes).as("range_start_bytes for %s.%s", keyspace, tableName).isNotNull(); + assertThat(rangeEndBytes).as("range_end_bytes for %s.%s", keyspace, tableName).isNotNull(); + + Range range = new Range(Long.parseLong(rangeStart), Long.parseLong(rangeEnd)); + Set exp = expected.get(range); + assertThat(exp).as("range (%s, %s] not in expected for %s", rangeStart, rangeEnd, keyspace) + .isNotNull(); + + Set expectedIps = exp.stream().map(e -> e.ip).collect(Collectors.toSet()); + Set expectedNodeIds = exp.stream().map(e -> e.nodeId).collect(Collectors.toSet()); + + assertThat(readEndpoints).as("read endpoints for %s.%s (%s,%s]", keyspace, tableName, rangeStart, rangeEnd) + .isEqualTo(expectedIps); + assertThat(writeEndpoints).as("write endpoints for %s.%s (%s,%s]", keyspace, tableName, rangeStart, rangeEnd) + .isEqualTo(expectedIps); + assertThat(readReplicas).as("read replica node-ids for %s.%s (%s,%s]", keyspace, tableName, rangeStart, rangeEnd) + .isEqualTo(expectedNodeIds); + assertThat(writeReplicas).as("write replica node-ids for %s.%s (%s,%s]", keyspace, tableName, rangeStart, rangeEnd) + .isEqualTo(expectedNodeIds); + + tablesFound.get(keyspace).add(tableName); + rangesFound.get(keyspace).add(range); + } + + assertThat(tablesFound.get(SIMPLE_KS)).as("tables in " + SIMPLE_KS).contains(SIMPLE_TABLE1, SIMPLE_TABLE2); + assertThat(rangesFound.get(SIMPLE_KS)).as("unique ranges in " + SIMPLE_KS).hasSize(SIMPLE_EXPECTED.size()); + assertThat(tablesFound.get(NTS_KS)).as("tables in " + NTS_KS).contains(NTS_TABLE1, NTS_TABLE2); + assertThat(rangesFound.get(NTS_KS)).as("unique ranges in " + NTS_KS).hasSize(NTS_EXPECTED.size()); + } + + @Test + @SuppressWarnings("unchecked") + public void testQueryByKeyspaceAndTable() + { + Object[][] result = CLUSTER.coordinator(1) + .execute("SELECT keyspace_name, table_name, range_start, range_end, " + + "read_endpoints, write_endpoints " + + "FROM system_views.data_placements WHERE keyspace_name = ? AND table_name = ?", + ConsistencyLevel.ONE, NTS_KS, NTS_TABLE1); + + assertThat(result).isNotEmpty(); + + Set rangesFound = new HashSet<>(); + for (Object[] row : result) + { + assertThat(row[0]).isEqualTo(NTS_KS); + assertThat(row[1]).isEqualTo(NTS_TABLE1); + + String rangeStart = (String) row[2]; + String rangeEnd = (String) row[3]; + Set readEndpoints = (Set) row[4]; + Set writeEndpoints = (Set) row[5]; + + Range range = new Range(Long.parseLong(rangeStart), Long.parseLong(rangeEnd)); + Set expected = NTS_EXPECTED.get(range); + assertThat(expected).as("range (%s, %s] not found in simulation", rangeStart, rangeEnd) + .isNotNull(); + + Set expectedIps = expected.stream().map(e -> e.ip).collect(Collectors.toSet()); + assertThat(readEndpoints).as("read endpoints for (%s,%s]", rangeStart, rangeEnd) + .isEqualTo(expectedIps); + assertThat(writeEndpoints).as("write endpoints for (%s,%s]", rangeStart, rangeEnd) + .isEqualTo(expectedIps); + + rangesFound.add(range); + } + + // The 2-component query must return ALL ranges for the table, not a subset. + assertThat(rangesFound).as("ranges for " + NTS_KS + "." + NTS_TABLE1) + .hasSize(NTS_EXPECTED.size()); + } + + @Test + @SuppressWarnings("unchecked") + public void testQueryByFullPartitionKey() + { + String rangeStart, rangeEnd; + { + Range range = NTS_EXPECTED.keySet().iterator().next(); + rangeStart = String.valueOf(range.start); + rangeEnd = String.valueOf(range.end); + } + + Object[][] result = CLUSTER.coordinator(1) + .execute("SELECT keyspace_name, table_name, range_start, range_end, " + + "read_endpoints, write_endpoints, read_replicas, write_replicas " + + "FROM system_views.data_placements " + + "WHERE keyspace_name = ? AND table_name = ? AND range_start = ? AND range_end = ?", + ConsistencyLevel.ONE, NTS_KS, NTS_TABLE1, rangeStart, rangeEnd); + + assertThat(result).as("full-PK query must return exactly one row").hasNumberOfRows(1); + + Object[] row = result[0]; + assertThat(row[0]).as("keyspace_name").isEqualTo(NTS_KS); + assertThat(row[1]).as("table_name").isEqualTo(NTS_TABLE1); + assertThat(row[2]).as("range_start").isEqualTo(rangeStart); + assertThat(row[3]).as("range_end").isEqualTo(rangeEnd); + + Set readEndpoints = (Set) row[4]; + Set writeEndpoints = (Set) row[5]; + Set readReplicas = (Set) row[6]; + Set writeReplicas = (Set) row[7]; + + Range range = new Range(Long.parseLong(rangeStart), Long.parseLong(rangeEnd)); + Set expected = NTS_EXPECTED.get(range); + assertThat(expected).as("range (%s, %s] not found in simulation", rangeStart, rangeEnd) + .isNotNull(); + + Set expectedIps = expected.stream().map(e -> e.ip).collect(Collectors.toSet()); + Set expectedNodeIds = expected.stream().map(e -> e.nodeId).collect(Collectors.toSet()); + + assertThat(readEndpoints).as("read endpoints").isEqualTo(expectedIps); + assertThat(writeEndpoints).as("write endpoints").isEqualTo(expectedIps); + assertThat(readReplicas).as("read replica node-ids").isEqualTo(expectedNodeIds); + assertThat(writeReplicas).as("write replica node-ids").isEqualTo(expectedNodeIds); + } + + @Shared + public static final class Range + { + public final long start; + public final long end; + + public Range(long start, long end) + { + this.start = start; + this.end = end; + } + + @Override + public boolean equals(Object o) + { + if (!(o instanceof Range)) return false; + Range r = (Range) o; + return r.start == start && r.end == end; + } + + @Override + public int hashCode() + { + return Long.hashCode(start) * 31 + Long.hashCode(end); + } + + @Override + public String toString() + { + return "(" + start + ", " + end + "]"; + } + } + + @Shared + public static final class Endpoint + { + public final String ip; + public final int nodeId; + + public Endpoint(String ip, int nodeId) + { + this.ip = ip; + this.nodeId = nodeId; + } + + @Override + public boolean equals(Object o) + { + if (!(o instanceof Endpoint)) return false; + Endpoint e = (Endpoint) o; + return e.nodeId == nodeId && e.ip.equals(ip); + } + + @Override + public int hashCode() + { + return ip.hashCode() * 31 + nodeId; + } + } + + private static Map> fetchExpectedPlacements(String keyspaceName) + { + return CLUSTER.get(1).callOnInstance(() -> { + Map> result = new HashMap<>(); + ClusterMetadata metadata = ClusterMetadata.current(); + KeyspaceMetadata ksm = metadata.schema.getKeyspaceMetadata(keyspaceName); + DataPlacement placement = metadata.placements.get(ksm.params.replication); + for (org.apache.cassandra.dht.Range range : placement.reads.ranges) + { + VersionedEndpoints.ForRange re = placement.reads.forRange(range); + Set endpoints = new HashSet<>(); + for (Replica r : re.get()) + { + NodeId nid = metadata.directory.peerId(r.endpoint()); + endpoints.add(new Endpoint(r.endpoint().toString(), nid.id())); + } + result.put(new Range(Long.parseLong(range.left.toString()), + Long.parseLong(range.right.toString())), + endpoints); + } + return result; + }); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/PartitionLocationTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/PartitionLocationTableTest.java new file mode 100644 index 000000000000..86c39dc6ca90 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/PartitionLocationTableTest.java @@ -0,0 +1,211 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Set; +import java.util.stream.Collectors; + +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ownership.DataPlacement; +import org.apache.cassandra.tcm.ownership.VersionedEndpoints; +import org.apache.cassandra.utils.Shared; + +import static org.assertj.core.api.Assertions.assertThat; + +public class PartitionLocationTableTest extends TestBaseImpl +{ + private static final int NUM_NODES = 3; + private static final String KEYSPACE = "test_ks"; + + private static Cluster CLUSTER; + + @BeforeClass + public static void setup() throws IOException + { + CLUSTER = Cluster.build(NUM_NODES).start(); + } + + @AfterClass + public static void cleanup() + { + if (CLUSTER != null) + CLUSTER.close(); + } + + + @Test + @SuppressWarnings("unchecked") + public void testSimplePartitionKey() + { + CLUSTER.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3}"); + CLUSTER.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl1 (col1 int PRIMARY KEY, col2 text)"); + + ExpectedLocation expected = computeExpectedLocation(KEYSPACE, "tbl1", "123"); + + Object[][] result = CLUSTER.coordinator(1).execute( + "SELECT tkn, range_start, range_end, range_start_bytes, range_end_bytes, " + + "read_endpoints, write_endpoints, read_replicas, write_replicas " + + "FROM system_views.partition_location " + + "WHERE keyspace_name = ? AND table_name = ? AND key = ?", + ConsistencyLevel.ONE, + KEYSPACE, "tbl1", "123"); + + Assert.assertEquals(1, result.length); + Object[] row = result[0]; + String token = (String) row[0]; + String rangeStart = (String) row[1]; + String rangeEnd = (String) row[2]; + ByteBuffer rangeStartBytes = (ByteBuffer) row[3]; + ByteBuffer rangeEndBytes = (ByteBuffer) row[4]; + Set readEndpoints = (Set) row[5]; + Set writeEndpoints = (Set) row[6]; + Set readReplicas = (Set) row[7]; + Set writeReplicas = (Set) row[8]; + + assertThat(token).as("token").isEqualTo(expected.token); + assertThat(rangeStart).as("range_start").isEqualTo(expected.rangeStart); + assertThat(rangeEnd).as("range_end").isEqualTo(expected.rangeEnd); + assertThat(rangeStartBytes).as("range_start_bytes").isNotNull(); + assertThat(rangeEndBytes).as("range_end_bytes").isNotNull(); + + assertThat(readEndpoints).as("read endpoints").isEqualTo(expected.readEndpoints); + assertThat(writeEndpoints).as("write endpoints").isEqualTo(expected.writeEndpoints); + assertThat(readReplicas).as("read replica node-ids").isEqualTo(expected.readReplicas); + assertThat(writeReplicas).as("write replica node-ids").isEqualTo(expected.writeReplicas); + } + + @Test + @SuppressWarnings("unchecked") + public void testCompositePartitionKey() + { + CLUSTER.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3}"); + CLUSTER.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl2 (col1 int, col2 text, col3 int, PRIMARY KEY ((col1, col2), col3))"); + + ExpectedLocation expected = computeExpectedLocation(KEYSPACE, "tbl2", "123:value1"); + + Object[][] result = CLUSTER.coordinator(1).execute( + "SELECT tkn, range_start, range_end, range_start_bytes, range_end_bytes, " + + "read_endpoints, write_endpoints, read_replicas, write_replicas " + + "FROM system_views.partition_location " + + "WHERE keyspace_name = ? AND table_name = ? AND key = ?", + ConsistencyLevel.ONE, + KEYSPACE, "tbl2", "123:value1"); + + Assert.assertEquals(1, result.length); + Object[] row = result[0]; + String token = (String) row[0]; + String rangeStart = (String) row[1]; + String rangeEnd = (String) row[2]; + ByteBuffer rangeStartBytes = (ByteBuffer) row[3]; + ByteBuffer rangeEndBytes = (ByteBuffer) row[4]; + Set readEndpoints = (Set) row[5]; + Set writeEndpoints = (Set) row[6]; + Set readReplicas = (Set) row[7]; + Set writeReplicas = (Set) row[8]; + + assertThat(token).as("token").isEqualTo(expected.token); + assertThat(rangeStart).as("range_start").isEqualTo(expected.rangeStart); + assertThat(rangeEnd).as("range_end").isEqualTo(expected.rangeEnd); + assertThat(rangeStartBytes).as("range_start_bytes").isNotNull(); + assertThat(rangeEndBytes).as("range_end_bytes").isNotNull(); + + assertThat(readEndpoints).as("read endpoints").isEqualTo(expected.readEndpoints); + assertThat(writeEndpoints).as("write endpoints").isEqualTo(expected.writeEndpoints); + assertThat(readReplicas).as("read replica node-ids").isEqualTo(expected.readReplicas); + assertThat(writeReplicas).as("write replica node-ids").isEqualTo(expected.writeReplicas); + } + + @Shared + public static final class ExpectedLocation // class needs to be public since it's shared + { + public final String token; + public final String rangeStart; + public final String rangeEnd; + public final Set readEndpoints; + public final Set writeEndpoints; + public final Set readReplicas; + public final Set writeReplicas; + + public ExpectedLocation(String token, String rangeStart, String rangeEnd, + Set readEndpoints, Set writeEndpoints, + Set readReplicas, Set writeReplicas) + { + this.token = token; + this.rangeStart = rangeStart; + this.rangeEnd = rangeEnd; + this.readEndpoints = readEndpoints; + this.writeEndpoints = writeEndpoints; + this.readReplicas = readReplicas; + this.writeReplicas = writeReplicas; + } + } + + private static ExpectedLocation computeExpectedLocation(String ks, String tbl, String key) + { + return CLUSTER.get(1).callOnInstance(() -> { + ClusterMetadata metadata = ClusterMetadata.current(); + KeyspaceMetadata ksm = metadata.schema.getKeyspaceMetadata(ks); + TableMetadata table = ksm.getTableOrViewNullable(tbl); + + ByteBuffer partitionKeyBytes = table.partitionKeyType.fromString(key); + DecoratedKey dk = table.partitioner.decorateKey(partitionKeyBytes); + Token token = dk.getToken(); + + DataPlacement placement = metadata.placements.get(ksm.params.replication); + VersionedEndpoints.ForRange readEps = placement.reads.forRange(token); + VersionedEndpoints.ForRange writeEps = placement.writes.forRange(token); + Range range = readEps.get().range(); + + Set readEndpoints = readEps.get().stream() + .map(r -> r.endpoint().toString()) + .collect(Collectors.toSet()); + Set writeEndpoints = writeEps.get().stream() + .map(r -> r.endpoint().toString()) + .collect(Collectors.toSet()); + Set readReplicas = readEps.get().stream() + .map(r -> metadata.directory.peerId(r.endpoint()).id()) + .collect(Collectors.toSet()); + Set writeReplicas = writeEps.get().stream() + .map(r -> metadata.directory.peerId(r.endpoint()).id()) + .collect(Collectors.toSet()); + + return new ExpectedLocation(token.toString(), + range.left.toString(), + range.right.toString(), + readEndpoints, + writeEndpoints, + readReplicas, + writeReplicas); + }); + } +}