在 Django PostgreSQL 中索引 JSONField答案

【问题标题】：Indexing JSONField in Django PostgreSQL在 Django PostgreSQL 中索引 JSONField
【发布时间】：2019-07-16 09:04:34
【问题描述】：

我正在使用一个简单的模型，该模型具有一个属性，该属性将该对象的所有数据存储在 JSONField 中。将其视为将 NoSQL 数据传输到我的 PostgreSQL 数据库的一种方式。有点像这样：

from django.contrib.postgres.fields import JSONField   

class Document(models.Model):
    content = JSONField()

每个Document 对象在其content 字段中（或多或少）具有相同的键，因此我使用这些键查询和排序这些文档。对于查询和排序，我使用的是 Django 的 annotate() 函数。我最近遇到了这个：

https://docs.djangoproject.com/en/2.1/ref/contrib/postgres/indexes/

我也知道 PostgreSQL 使用 JSONB，这显然是可索引的。所以我的问题是：我可以索引我的content 字段以使我的读取操作更快地处理复杂的查询吗？如果是这样，那我该怎么做？我链接的文档页面没有示例。

【问题讨论】：

我认为有与此相关的类似问题 - stackoverflow.com/a/49358119/4116955 和 stackoverflow.com/questions/17807030/… 请使用其 JSONB 类型将您的 json 数据迁移到 postgresql 表，然后尝试索引整个列或特定字段json 有效负载。
这很有趣。但是，我正在寻找一种在 Django 中执行此操作的方法，因为这将使我的代码库更易于管理。
这样的东西应该可以工作 - class Doc(models.Model): data = JSONField() class Meta: indexes = [ GinIndex( fields=['data'], name='data_gin', ), ] 抱歉格式错误。
这不会索引整个列吗？考虑到 JSONField 中可能存在各种数据类型，这是一个好主意吗？
我认为当您查询表中 JSONB 字段中的键映射到特定值的行时，此索引将起作用

标签： django postgresql

【解决方案1】：

对于那些想要索引特定键的人，创建原始 sql 迁移：

运行./manage.py makemigrations --empty yourApp，其中yourApp 是您要为其更改索引的模型的应用程序。
编辑迁移，即

operations = [
    migrations.RunSQL("CREATE INDEX idx_name ON your_table((json_field->'json_key'));")
]

idx_name 是索引的名称，your_table 是您的表，json_field 是您的 JSONField，json_key 在这种情况下是您要索引的键。

注意索引创建代码行中的-> 运算符，而不是here 中提到的->>。

应该这样做，但要验证一切顺利，请运行以下 sql：

SELECT
    indexname,
    indexdef
FROM
    pg_indexes
WHERE
    tablename = '<your-table>';

看看你的索引是否在那里。

【讨论】：

我认为在 3.2 中添加了原生支持作为此 PR 的一部分：github.com/django/django/pull/11929 但我不清楚如何使用。所有请求 JSONB 字段索引的 Django 票证都基于该 PR 关闭。

【解决方案2】：

还有一种更通用的 Django 原生方式。您可以使用以下custom Migration Operation：

class CreateJsonbObjectKeyIndex(Operation):

    reversible = True

    def __init__(self, model_name, field, key, index_type='btree', concurrently=False, name=None):
        self.model_name = model_name
        self.field = field
        self.key = key
        self.index_type = index_type
        self.concurrently = concurrently
        self.name = name

    def state_forwards(self, app_label, state):
        pass

    def get_names(self, app_label, schema_editor, from_state, to_state):
        table_name = from_state.apps.get_model(app_label, self.model_name)._meta.db_table
        index_name = schema_editor.quote_name(
            self.name or schema_editor._create_index_name(table_name, [f'{self.field}__{self.key}'])
        )
        return table_name, index_name

    def database_forwards(self, app_label, schema_editor, from_state, to_state):
        table_name, index_name = self.get_names(app_label, schema_editor, from_state, to_state)
        schema_editor.execute(f"""
            CREATE INDEX {'CONCURRENTLY' if self.concurrently else ''} {index_name} 
            ON {table_name}
            USING {self.index_type}
            (({self.field}->'{self.key}'));
        """)

    def database_backwards(self, app_label, schema_editor, from_state, to_state):
        _, index_name = self.get_names(app_label, schema_editor, from_state, to_state)
        schema_editor.execute(f"DROP INDEX {index_name};")

    def describe(self):
        return f'Creates index for JSONB object field {self.field}->{self.key} of {self.model_name} model'

    @property
    def migration_name_fragment(self):
        return f'create_index_{self.model_name}_{self.field}_{self.key}'

使用示例：

from django.db import migrations

from util.migration import CreateJsonbObjectKeyIndex


class Migration(migrations.Migration):
    atomic = False  # Required if concurrently=True for 0 downtime background index creation

    dependencies = [
        ('app_label', '00XX_prev_migration'),
    ]

    operations = [
        migrations.SeparateDatabaseAndState(
            database_operations=[
                # Operation to run custom SQL command. Check the output of `sqlmigrate` to see the auto-generated SQL
                CreateJsonbObjectKeyIndex(
                    model_name='User', field='meta', key='adid', index_type='HASH',
                    concurrently=True,
                )
            ],
        )
    ]

使用 Django-2.2 和 AWS Postgres RDS 测试，但应该与其他 Django 兼容

【讨论】：