Repository: getmoto/moto (python)

Issue:
DynamoDB sorts by Table hash before GSI when comparing ExclusiveStartKey
When querying using a GSI for items that sort in one direction by the GSI's sort key and in the opposite direction by the table's hash key it appears that the hash key sort takes precedence when comparing whether the item has already been seen (comparing with the ExclusiveStartKey).

It appears that this comes from https://github.com/getmoto/moto/blob/1c8c0fe4d9c7f5481a40814a8a6459a5142707ff/moto/dynamodb/models/table.py#L1035 the change around which was introduced in https://github.com/getmoto/moto/pull/7795

A test which demonstrates this is below, the 2 items are sorted in one direction by their table's hash key, but in the opposite direction by their GSI key (same hash value, different sort values).

```
@mock_aws
def test_query_gsi_pagination_with_string_gsi_range_with_opposite_sort():
    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")

    # Create the DynamoDB table.
    dynamodb.create_table(
        TableName="users",
        KeySchema=[
            {"AttributeName": "pk", "KeyType": "HASH"},
        ],
        AttributeDefinitions=[
            {"AttributeName": "pk", "AttributeType": "S"},
            {"AttributeName": "gsi_hash", "AttributeType": "S"},
            {"AttributeName": "gsi_sort", "AttributeType": "S"},
        ],
        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
        GlobalSecondaryIndexes=[
            {
                "IndexName": "gsi",
                "KeySchema": [
                    {"AttributeName": "gsi_hash", "KeyType": "HASH"},
                    {"AttributeName": "gsi_sort", "KeyType": "RANGE"},
                ],
                "Projection": {"ProjectionType": "ALL"},
            }
        ],
    )
    table = dynamodb.Table("users")

    table.put_item(Item={"pk": "b", "gsi_hash": "a", "gsi_sort": "a"})
    table.put_item(Item={"pk": "a", "gsi_hash": "a", "gsi_sort": "b"})

    page1 = table.query(
        KeyConditionExpression=Key("gsi_hash").eq("a"),
        IndexName="gsi",
        Limit=1,
    )
    assert page1["Count"] == 1
    assert page1["ScannedCount"] == 1
    assert len(page1["Items"]) == 1

    page2 = table.query(
        KeyConditionExpression=Key("gsi_hash").eq("a"),
        IndexName="gsi",
        Limit=1,
        ExclusiveStartKey=page1["LastEvaluatedKey"],
    )
    assert page2["Count"] == 1 # <------------------------------ fails here 
    assert page2["ScannedCount"] == 1
    assert len(page2["Items"]) == 1
    assert "LastEvaluatedKey" not in page2

    results = page1["Items"] + page2["Items"]
    subjects = set([(r["pk"]) for r in results])
    assert subjects == ["b", "a"]
```

It seems the logic should be:
1. Sort by GSI hash key if present
2. If equal sort by GSI sort key
3. If equal sort by table hash key
4. If equal sort by table sort key

Python version 3.11
Latest moto (cloned repo and added test into tests directly, hoping I could figure out a fix if my understanding of the expected behaviour was right)


Pull request description:
DynamoDB: Sorts by the keys of the GSI before sorting by keys of the table
Fixes https://github.com/getmoto/moto/issues/7834

If any of the keys don't exist (because we're not using a GSI or the item doesn't have them) falls through to the next attribute in the list.

> This is my first PR contributing to moto - please let me know on any standards/etc. that I've missed (ran tests and lint locally, both were passing)

Patch:
diff --git a/moto/dynamodb/models/table.py b/moto/dynamodb/models/table.py
index 2696608fe..244fb2fa4 100644
--- a/moto/dynamodb/models/table.py
+++ b/moto/dynamodb/models/table.py
@@ -1026,39 +1026,33 @@ class Table(CloudFormationModel):
         If that is the case, we compare by the RK of the main table instead
         Related: https://github.com/getmoto/moto/issues/7761
         """
-        hash_comes_before = any(
-            item.attrs[hash_key_attr] != DynamoType(dct.get(hash_key_attr))  # type: ignore
-            for hash_key_attr in hash_key_attrs
-            if hash_key_attr in item.attrs
+        gsi_hash_key = hash_key_attrs[0] if len(hash_key_attrs) == 2 else None
+        table_hash_key = str(
+            hash_key_attrs[0] if gsi_hash_key is None else hash_key_attrs[1]
         )
-        if hash_comes_before:
-            # If hash keys are different, order immediately
-            return (
-                any(
-                    item.attrs[hash_key_attr] < DynamoType(dct.get(hash_key_attr))  # type: ignore
-                    for hash_key_attr in hash_key_attrs
-                    if hash_key_attr in item.attrs
-                )
-                == scan_index_forward
-            )
-        if not any(range_key_attrs):
-            # If hash keys match and no range key, items are identical
-            return True
-        # We know hash keys are equal, use range key as tiebreaker
-        if all(
-            item.attrs[attr] == DynamoType(dct.get(attr))  # type: ignore[arg-type]
-            for attr in range_key_attrs
-            if attr
-        ):
-            # If range keys (also) match, items are identical
-            return True
-        # Range keys are different, order accordingly
-        comes_before = any(
-            item.attrs[attr] < DynamoType(dct.get(attr))  # type: ignore[arg-type]
-            for attr in range_key_attrs
-            if attr
+        gsi_range_key = range_key_attrs[0] if len(range_key_attrs) == 2 else None
+        table_range_key = str(
+            range_key_attrs[0] if gsi_range_key is None else range_key_attrs[1]
         )
-        return comes_before == scan_index_forward
+        # Gets the GSI and table hash and range keys in the order to try sorting by
+        attrs_to_sort_by = [
+            gsi_hash_key,
+            gsi_range_key,
+            table_hash_key,
+            table_range_key,
+        ]
+        for attr in attrs_to_sort_by:
+            if (
+                attr is not None
+                and attr in item.attrs
+                and item.attrs[attr] != DynamoType(dct.get(attr))  # type: ignore
+            ):
+                return (
+                    (item.attrs[attr] < DynamoType(dct.get(attr)))  # type: ignore
+                    == scan_index_forward
+                )
+        # Keys were equal, items are identical
+        return True
 
     def _get_last_evaluated_key(
         self, last_result: Item, index_name: Optional[str]


Test patch:
diff --git a/tests/test_dynamodb/test_dynamodb_query.py b/tests/test_dynamodb/test_dynamodb_query.py
index 4770efd3e..c5619a3ad 100644
--- a/tests/test_dynamodb/test_dynamodb_query.py
+++ b/tests/test_dynamodb/test_dynamodb_query.py
@@ -326,6 +326,63 @@ def test_query_gsi_pagination_with_string_gsi_range(table_name=None):
     assert subjects == set(range(10))
 
 
+@mock_aws
+def test_query_gsi_pagination_with_opposite_pk_order():
+    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
+
+    # Create the DynamoDB table.
+    dynamodb.create_table(
+        TableName="users",
+        KeySchema=[
+            {"AttributeName": "pk", "KeyType": "HASH"},
+        ],
+        AttributeDefinitions=[
+            {"AttributeName": "pk", "AttributeType": "S"},
+            {"AttributeName": "gsi_hash", "AttributeType": "S"},
+            {"AttributeName": "gsi_sort", "AttributeType": "S"},
+        ],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+        GlobalSecondaryIndexes=[
+            {
+                "IndexName": "gsi",
+                "KeySchema": [
+                    {"AttributeName": "gsi_hash", "KeyType": "HASH"},
+                    {"AttributeName": "gsi_sort", "KeyType": "RANGE"},
+                ],
+                "Projection": {"ProjectionType": "ALL"},
+            }
+        ],
+    )
+    table = dynamodb.Table("users")
+
+    table.put_item(Item={"pk": "b", "gsi_hash": "a", "gsi_sort": "a"})
+    table.put_item(Item={"pk": "a", "gsi_hash": "a", "gsi_sort": "b"})
+
+    page1 = table.query(
+        KeyConditionExpression=Key("gsi_hash").eq("a"),
+        IndexName="gsi",
+        Limit=1,
+    )
+    assert page1["Count"] == 1
+    assert page1["ScannedCount"] == 1
+    assert len(page1["Items"]) == 1
+
+    page2 = table.query(
+        KeyConditionExpression=Key("gsi_hash").eq("a"),
+        IndexName="gsi",
+        Limit=1,
+        ExclusiveStartKey=page1["LastEvaluatedKey"],
+    )
+    assert page2["Count"] == 1
+    assert page2["ScannedCount"] == 1
+    assert len(page2["Items"]) == 1
+    assert "LastEvaluatedKey" not in page2
+
+    results = page1["Items"] + page2["Items"]
+    subjects = [(r["pk"]) for r in results]
+    assert subjects == ["b", "a"]
+
+
 @pytest.mark.aws_verified
 @dynamodb_aws_verified(add_range=True, add_gsi_range=True)
 def test_query_gsi_pagination_with_string_gsi_range_and_empty_gsi_pk(table_name=None):