Uploaded image for project: 'Cassandra'
  1. Cassandra
  2. CASSANDRA-15637

CqlInputFormat regression going from 2.1 to 3.x caused by semantic difference between thrift and the new system.size_estimates table when dealing with multiple dc deployments

    XMLWordPrintableJSON

    Details

    • Bug Category:
      Correctness - API / Semantic Implementation
    • Severity:
      Critical
    • Complexity:
      Challenging
    • Discovered By:
      User Report
    • Platform:
      All
    • Impacts:
      None
    • Since Version:
    • Test and Documentation Plan:
      Hide

      tested cluster using MapReduce. The test case targeted is in the description

      Show
      tested cluster using MapReduce. The test case targeted is in the description

      Description

      In 3.0 CqlInputFormat switched away from thrift in favor of a new system.size_estimates table, but the semantics changed when dealing with multiple DCs or when Cassandra is not collocated with Hadoop.

      The core issues are:

      • system.size_estimates uses the primary range, in a multi-dc setup this could lead to uneven ranges
        example:
      DC1: [0, 10, 20, 30]
      DC2: [1, 11, 21, 31]
      DC3: [2, 12, 22, 32]
      

      Using NetworkTopologyStrategy the primary ranges are: [0, 1), [1, 2), [2, 10), [10, 11), [11, 12), [12, 20), [20, 21), [21, 22), [22, 30), [30, 31), [31, 32), [32, 0).
      Given this the only ranges that are more than one token are: [2, 10), [12, 20), [22, 30).

      • system.size_estimates is not replicated so need to hit every node in the cluster to get estimates, if nodes are down in the DC with non-size-1 ranges there is no way to get a estimate.
      • CqlInputFormat used to call describe_local_ring so all interactions were with a single DC, the java driver doesn't filter the DC so looks to allow cross DC traffic and includes nodes from other DCs in the replica set; in the example above, the amount of splits went from 4 to 12.
      • CqlInputFormat used to call describe_splits_ex to dynamically calculate the estimates, this was on the "local primary range" and was able to hit replicas to create estimates if the primary was down. With system.size_estimates we no longer have backup and no longer expose the "local primary range" in multi-dc.
      • CqlInputFormat had a config cassandra.input.keyRange which let you define your own range. If the range doesn't perfectly match the local range then the intersectWith calls will produce ranges with no estimates. Example: [0, 10, 20], cassandra.input.keyRange=5,15. This won't find any estimates so will produce 2 splits with 128 estimate (default when not found).
      • CqlInputFormat special cases Cassandra being collocated with Hadoop and assumes this when querying system.size_estimates as it doesn't filter to the specific host, this means that non-collocated deployments randomly select the nodes and create splits with ranges the hosts do not have locally.

      The problems are deterministic to replicate, the following test will show it

      1) deploy a 3 DC cluster with 3 nodes each
      2) create DC2 tokens are +1 of DC1 and DC3 are +1 of DC2
      3) CREATE KEYSPACE simpleuniform0 WITH replication =

      {‘class’: ‘NetworkTopologyStrategy’, ‘DC1’: 3, ‘DC2’: 3, ‘DC3’: 3}

      ;
      4) CREATE TABLE simpletable0 (pk bigint, ck bigint, value blob, PRIMARY KEY (pk, ck))
      5) insert 500k partitions uniformly: [0, 500,000)
      6) wait until estimates catch up to writes
      7) for all nodes, SELECT * FROM system.size_estimates

      You will get the following

       keyspace_name  | table_name   | range_start          | range_end            | mean_partition_size | partitions_count
      ----------------+--------------+----------------------+----------------------+---------------------+------------------
       simpleuniform0 | simpletable0 | -9223372036854775808 | -6148914691236517206 |                  87 |           122240
       simpleuniform0 | simpletable0 |  6148914691236517207 | -9223372036854775808 |                  87 |           121472
      
      (2 rows)
      
       keyspace_name  | table_name   | range_start | range_end           | mean_partition_size | partitions_count
      ----------------+--------------+-------------+---------------------+---------------------+------------------
       simpleuniform0 | simpletable0 |           2 | 6148914691236517205 |                  87 |           243072
      
      (1 rows)
      
       keyspace_name  | table_name   | range_start          | range_end            | mean_partition_size | partitions_count
      ----------------+--------------+----------------------+----------------------+---------------------+------------------
       simpleuniform0 | simpletable0 | -6148914691236517206 | -6148914691236517205 |                  87 |                1
      
      (1 rows)
      
       keyspace_name  | table_name   | range_start | range_end | mean_partition_size | partitions_count
      ----------------+--------------+-------------+-----------+---------------------+------------------
       simpleuniform0 | simpletable0 |           0 |         1 |                  87 |                1
      
      (1 rows)
      
       keyspace_name  | table_name   | range_start         | range_end           | mean_partition_size | partitions_count
      ----------------+--------------+---------------------+---------------------+---------------------+------------------
       simpleuniform0 | simpletable0 | 6148914691236517205 | 6148914691236517206 |                  87 |                1
      
      (1 rows)
      
       keyspace_name  | table_name   | range_start          | range_end            | mean_partition_size | partitions_count
      ----------------+--------------+----------------------+----------------------+---------------------+------------------
       simpleuniform0 | simpletable0 | -6148914691236517205 | -6148914691236517204 |                  87 |                1
      
      (1 rows)
      
       keyspace_name  | table_name   | range_start | range_end | mean_partition_size | partitions_count
      ----------------+--------------+-------------+-----------+---------------------+------------------
       simpleuniform0 | simpletable0 |           1 |         2 |                  87 |                1
      
      (1 rows)
      
       keyspace_name  | table_name   | range_start         | range_end           | mean_partition_size | partitions_count
      ----------------+--------------+---------------------+---------------------+---------------------+------------------
       simpleuniform0 | simpletable0 | 6148914691236517206 | 6148914691236517207 |                  87 |                1
      
      (1 rows)
      

      8) create a MR job against simpleuniform0. simpletable0, you will get 10 splits where as 2.1 was 4

        Attachments

          Activity

            People

            • Assignee:
              dcapwell David Capwell
              Reporter:
              dcapwell David Capwell
              Authors:
              David Capwell
              Reviewers:
              Aleksey Yeschenko, Brandon Williams
            • Votes:
              0 Vote for this issue
              Watchers:
              4 Start watching this issue

              Dates

              • Created:
                Updated:
                Resolved:

                Time Tracking

                Estimated:
                Original Estimate - Not Specified
                Not Specified
                Remaining:
                Remaining Estimate - 0h
                0h
                Logged:
                Time Spent - 10m
                10m