class SqlContentEntityStorage

class SqlContentEntityStorage extends ContentEntityStorageBase implements SqlEntityStorageInterface, DynamicallyFieldableEntityStorageSchemaInterface, EntityBundleListenerInterface {
    
    /**
     * The entity type's field storage definitions.
     *
     * @var \Drupal\Core\Field\FieldStorageDefinitionInterface[]
     */
    protected $fieldStorageDefinitions;
    
    /**
     * The mapping of field columns to SQL tables.
     *
     * @var \Drupal\Core\Entity\Sql\TableMappingInterface
     */
    protected $tableMapping;
    
    /**
     * Name of entity's revision database table field, if it supports revisions.
     *
     * Has the value FALSE if this entity does not use revisions.
     *
     * @var string
     */
    protected $revisionKey = FALSE;
    
    /**
     * The entity langcode key.
     *
     * @var string|bool
     */
    protected $langcodeKey = FALSE;
    
    /**
     * The default language entity key.
     *
     * @var string
     */
    protected $defaultLangcodeKey = FALSE;
    
    /**
     * The base table of the entity.
     *
     * @var string
     */
    protected $baseTable;
    
    /**
     * The table that stores revisions, if the entity supports revisions.
     *
     * @var string
     */
    protected $revisionTable;
    
    /**
     * The table that stores properties, if the entity has multilingual support.
     *
     * @var string
     */
    protected $dataTable;
    
    /**
     * The table that stores revision field data if the entity supports revisions.
     *
     * @var string
     */
    protected $revisionDataTable;
    
    /**
     * Active database connection.
     *
     * @var \Drupal\Core\Database\Connection
     */
    protected $database;
    
    /**
     * The entity type's storage schema object.
     *
     * @var \Drupal\Core\Entity\Schema\EntityStorageSchemaInterface
     */
    protected $storageSchema;
    
    /**
     * The language manager.
     *
     * @var \Drupal\Core\Language\LanguageManagerInterface
     */
    protected $languageManager;
    
    /**
     * The entity type manager.
     *
     * @var \Drupal\Core\Entity\EntityTypeManagerInterface
     */
    protected $entityTypeManager;
    
    /**
     * Whether this storage should use the temporary table mapping.
     *
     * @var bool
     */
    protected $temporary = FALSE;
    
    /**
     * {@inheritdoc}
     */
    public static function createInstance(ContainerInterface $container, EntityTypeInterface $entity_type) {
        return new static($entity_type, $container->get('database'), $container->get('entity_field.manager'), $container->get('cache.entity'), $container->get('language_manager'), $container->get('entity.memory_cache'), $container->get('entity_type.bundle.info'), $container->get('entity_type.manager'));
    }
    
    /**
     * Constructs a SqlContentEntityStorage object.
     *
     * @param \Drupal\Core\Entity\EntityTypeInterface $entity_type
     *   The entity type definition.
     * @param \Drupal\Core\Database\Connection $database
     *   The database connection to be used.
     * @param \Drupal\Core\Entity\EntityFieldManagerInterface $entity_field_manager
     *   The entity field manager.
     * @param \Drupal\Core\Cache\CacheBackendInterface $cache
     *   The cache backend to be used.
     * @param \Drupal\Core\Language\LanguageManagerInterface $language_manager
     *   The language manager.
     * @param \Drupal\Core\Cache\MemoryCache\MemoryCacheInterface $memory_cache
     *   The memory cache backend to be used.
     * @param \Drupal\Core\Entity\EntityTypeBundleInfoInterface $entity_type_bundle_info
     *   The entity type bundle info.
     * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entity_type_manager
     *   The entity type manager.
     */
    public function __construct(EntityTypeInterface $entity_type, Connection $database, EntityFieldManagerInterface $entity_field_manager, CacheBackendInterface $cache, LanguageManagerInterface $language_manager, MemoryCacheInterface $memory_cache, EntityTypeBundleInfoInterface $entity_type_bundle_info, EntityTypeManagerInterface $entity_type_manager) {
        parent::__construct($entity_type, $entity_field_manager, $cache, $memory_cache, $entity_type_bundle_info);
        $this->database = $database;
        $this->languageManager = $language_manager;
        $this->entityTypeManager = $entity_type_manager;
        $this->entityType = $this->entityTypeManager
            ->getActiveDefinition($entity_type->id());
        $this->fieldStorageDefinitions = $this->entityFieldManager
            ->getActiveFieldStorageDefinitions($entity_type->id());
        $this->initTableLayout();
    }
    
    /**
     * Initializes table name variables.
     */
    protected function initTableLayout() {
        // Reset table field values to ensure changes in the entity type definition
        // are correctly reflected in the table layout.
        $this->tableMapping = NULL;
        $this->revisionKey = NULL;
        $this->revisionTable = NULL;
        $this->dataTable = NULL;
        $this->revisionDataTable = NULL;
        $table_mapping = $this->getTableMapping();
        $this->baseTable = $table_mapping->getBaseTable();
        $revisionable = $this->entityType
            ->isRevisionable();
        if ($revisionable) {
            $this->revisionKey = $this->entityType
                ->getKey('revision') ?: 'revision_id';
            $this->revisionTable = $table_mapping->getRevisionTable();
        }
        $translatable = $this->entityType
            ->isTranslatable();
        if ($translatable) {
            $this->dataTable = $table_mapping->getDataTable();
            $this->langcodeKey = $this->entityType
                ->getKey('langcode');
            $this->defaultLangcodeKey = $this->entityType
                ->getKey('default_langcode');
        }
        if ($revisionable && $translatable) {
            $this->revisionDataTable = $table_mapping->getRevisionDataTable();
        }
    }
    
    /**
     * Gets the base table name.
     *
     * @return string
     *   The table name.
     */
    public function getBaseTable() {
        return $this->baseTable;
    }
    
    /**
     * Gets the revision table name.
     *
     * @return string|false
     *   The table name or FALSE if it is not available.
     */
    public function getRevisionTable() {
        return $this->revisionTable;
    }
    
    /**
     * Gets the data table name.
     *
     * @return string|false
     *   The table name or FALSE if it is not available.
     */
    public function getDataTable() {
        return $this->dataTable;
    }
    
    /**
     * Gets the revision data table name.
     *
     * @return string|false
     *   The table name or FALSE if it is not available.
     */
    public function getRevisionDataTable() {
        return $this->revisionDataTable;
    }
    
    /**
     * Gets the entity type's storage schema object.
     *
     * @return \Drupal\Core\Entity\Sql\SqlContentEntityStorageSchema
     *   The schema object.
     */
    protected function getStorageSchema() {
        if (!isset($this->storageSchema)) {
            $class = $this->entityType
                ->getHandlerClass('storage_schema') ?: 'Drupal\\Core\\Entity\\Sql\\SqlContentEntityStorageSchema';
            $this->storageSchema = new $class($this->entityTypeManager, $this->entityType, $this, $this->database, $this->entityFieldManager);
        }
        return $this->storageSchema;
    }
    
    /**
     * Updates the wrapped entity type definition.
     *
     * @param \Drupal\Core\Entity\EntityTypeInterface $entity_type
     *   The update entity type.
     *
     * @internal Only to be used internally by Entity API. Expected to be
     *   removed by https://www.drupal.org/node/2274017.
     */
    public function setEntityType(EntityTypeInterface $entity_type) {
        if ($this->entityType
            ->id() == $entity_type->id()) {
            $this->entityType = $entity_type;
            $this->initTableLayout();
        }
        else {
            throw new EntityStorageException("Unsupported entity type {$entity_type->id()}");
        }
    }
    
    /**
     * Updates the internal list of field storage definitions.
     *
     * @param \Drupal\Core\Field\FieldStorageDefinitionInterface[] $field_storage_definitions
     *   An array of field storage definitions.
     *
     * @internal Only to be used internally by Entity API.
     */
    public function setFieldStorageDefinitions(array $field_storage_definitions) {
        foreach ($field_storage_definitions as $field_storage_definition) {
            if ($field_storage_definition->getTargetEntityTypeId() !== $this->entityType
                ->id()) {
                throw new EntityStorageException("Unsupported entity type {$field_storage_definition->getTargetEntityTypeId()}");
            }
        }
        $this->fieldStorageDefinitions = $field_storage_definitions;
    }
    
    /**
     * Sets the wrapped table mapping definition.
     *
     * @param \Drupal\Core\Entity\Sql\TableMappingInterface $table_mapping
     *   The table mapping.
     *
     * @internal Only to be used internally by Entity API. Expected to be removed
     *   by https://www.drupal.org/node/2554235.
     */
    public function setTableMapping(TableMappingInterface $table_mapping) {
        $this->tableMapping = $table_mapping;
        $this->baseTable = $table_mapping->getBaseTable();
        $this->revisionTable = $table_mapping->getRevisionTable();
        $this->dataTable = $table_mapping->getDataTable();
        $this->revisionDataTable = $table_mapping->getRevisionDataTable();
    }
    
    /**
     * Changes the temporary state of the storage.
     *
     * @param bool $temporary
     *   Whether to use a temporary table mapping or not.
     *
     * @internal Only to be used internally by Entity API.
     */
    public function setTemporary($temporary) {
        $this->temporary = $temporary;
    }
    
    /**
     * {@inheritdoc}
     */
    public function getTableMapping(array $storage_definitions = NULL) {
        // If a new set of field storage definitions is passed, for instance when
        // comparing old and new storage schema, we compute the table mapping
        // without caching.
        if ($storage_definitions) {
            return $this->getCustomTableMapping($this->entityType, $storage_definitions);
        }
        // If we are using our internal storage definitions, which is our main use
        // case, we can statically cache the computed table mapping.
        if (!isset($this->tableMapping)) {
            $this->tableMapping = $this->getCustomTableMapping($this->entityType, $this->fieldStorageDefinitions);
        }
        return $this->tableMapping;
    }
    
    /**
     * Gets a table mapping for the specified entity type and storage definitions.
     *
     * @param \Drupal\Core\Entity\ContentEntityTypeInterface $entity_type
     *   An entity type definition.
     * @param \Drupal\Core\Field\FieldStorageDefinitionInterface[] $storage_definitions
     *   An array of field storage definitions to be used to compute the table
     *   mapping.
     * @param string $prefix
     *   (optional) A prefix to be used by all the tables of this mapping.
     *   Defaults to an empty string.
     *
     * @return \Drupal\Core\Entity\Sql\TableMappingInterface
     *   A table mapping object for the entity's tables.
     *
     * @internal
     */
    public function getCustomTableMapping(ContentEntityTypeInterface $entity_type, array $storage_definitions, $prefix = '') {
        $prefix = $prefix ?: ($this->temporary ? 'tmp_' : '');
        return DefaultTableMapping::create($entity_type, $storage_definitions, $prefix);
    }
    
    /**
     * {@inheritdoc}
     */
    protected function doLoadMultiple(array $ids = NULL) {
        // Attempt to load entities from the persistent cache. This will remove IDs
        // that were loaded from $ids.
        $entities_from_cache = $this->getFromPersistentCache($ids);
        // Load any remaining entities from the database.
        if ($entities_from_storage = $this->getFromStorage($ids)) {
            $this->invokeStorageLoadHook($entities_from_storage);
            $this->setPersistentCache($entities_from_storage);
        }
        return $entities_from_cache + $entities_from_storage;
    }
    
    /**
     * Gets entities from the storage.
     *
     * @param array|null $ids
     *   If not empty, return entities that match these IDs. Return all entities
     *   when NULL.
     *
     * @return \Drupal\Core\Entity\ContentEntityInterface[]
     *   Array of entities from the storage.
     */
    protected function getFromStorage(array $ids = NULL) {
        $entities = [];
        if (!empty($ids)) {
            // Sanitize IDs. Before feeding ID array into buildQuery, check whether
            // it is empty as this would load all entities.
            $ids = $this->cleanIds($ids);
        }
        if ($ids === NULL || $ids) {
            // Build and execute the query.
            $query_result = $this->buildQuery($ids)
                ->execute();
            $records = $query_result->fetchAllAssoc($this->idKey);
            // Map the loaded records into entity objects and according fields.
            if ($records) {
                $entities = $this->mapFromStorageRecords($records);
            }
        }
        return $entities;
    }
    
    /**
     * Maps from storage records to entity objects, and attaches fields.
     *
     * @param array $records
     *   Associative array of query results, keyed on the entity ID or revision
     *   ID.
     * @param bool $load_from_revision
     *   (optional) Flag to indicate whether revisions should be loaded or not.
     *   Defaults to FALSE.
     *
     * @return array
     *   An array of entity objects implementing the EntityInterface.
     */
    protected function mapFromStorageRecords(array $records, $load_from_revision = FALSE) {
        if (!$records) {
            return [];
        }
        // Get the names of the fields that are stored in the base table and, if
        // applicable, the revision table. Other entity data will be loaded in
        // loadFromSharedTables() and loadFromDedicatedTables().
        $field_names = $this->tableMapping
            ->getFieldNames($this->baseTable);
        if ($this->revisionTable) {
            $field_names = array_unique(array_merge($field_names, $this->tableMapping
                ->getFieldNames($this->revisionTable)));
        }
        $values = [];
        foreach ($records as $id => $record) {
            $values[$id] = [];
            // Skip the item delta and item value levels (if possible) but let the
            // field assign the value as suiting. This avoids unnecessary array
            // hierarchies and saves memory here.
            foreach ($field_names as $field_name) {
                $field_columns = $this->tableMapping
                    ->getColumnNames($field_name);
                // Handle field types that store several properties.
                if (count($field_columns) > 1) {
                    $definition_columns = $this->fieldStorageDefinitions[$field_name]
                        ->getColumns();
                    foreach ($field_columns as $property_name => $column_name) {
                        if (property_exists($record, $column_name)) {
                            $values[$id][$field_name][LanguageInterface::LANGCODE_DEFAULT][$property_name] = !empty($definition_columns[$property_name]['serialize']) ? unserialize($record->{$column_name}) : $record->{$column_name};
                            unset($record->{$column_name});
                        }
                    }
                }
                else {
                    $column_name = reset($field_columns);
                    if (property_exists($record, $column_name)) {
                        $columns = $this->fieldStorageDefinitions[$field_name]
                            ->getColumns();
                        $column = reset($columns);
                        $values[$id][$field_name][LanguageInterface::LANGCODE_DEFAULT] = !empty($column['serialize']) ? unserialize($record->{$column_name}) : $record->{$column_name};
                        unset($record->{$column_name});
                    }
                }
            }
            // Handle additional record entries that are not provided by an entity
            // field, such as 'isDefaultRevision'.
            foreach ($record as $name => $value) {
                $values[$id][$name][LanguageInterface::LANGCODE_DEFAULT] = $value;
            }
        }
        // Initialize translations array.
        $translations = array_fill_keys(array_keys($values), []);
        // Load values from shared and dedicated tables.
        $this->loadFromSharedTables($values, $translations, $load_from_revision);
        $this->loadFromDedicatedTables($values, $load_from_revision);
        $entities = [];
        foreach ($values as $id => $entity_values) {
            $bundle = $this->bundleKey ? $entity_values[$this->bundleKey][LanguageInterface::LANGCODE_DEFAULT] : NULL;
            // Turn the record into an entity class.
            $entity_class = $this->getEntityClass($bundle);
            $entities[$id] = new $entity_class($entity_values, $this->entityTypeId, $bundle, array_keys($translations[$id]));
        }
        return $entities;
    }
    
    /**
     * Loads values for fields stored in the shared data tables.
     *
     * @param array &$values
     *   Associative array of entities values, keyed on the entity ID or the
     *   revision ID.
     * @param array &$translations
     *   List of translations, keyed on the entity ID.
     * @param bool $load_from_revision
     *   Flag to indicate whether revisions should be loaded or not.
     */
    protected function loadFromSharedTables(array &$values, array &$translations, $load_from_revision) {
        $record_key = !$load_from_revision ? $this->idKey : $this->revisionKey;
        if ($this->dataTable) {
            // If a revision table is available, we need all the properties of the
            // latest revision. Otherwise we fall back to the data table.
            $table = $this->revisionDataTable ?: $this->dataTable;
            $alias = $this->revisionDataTable ? 'revision' : 'data';
            $query = $this->database
                ->select($table, $alias, [
                'fetch' => \PDO::FETCH_ASSOC,
            ])
                ->fields($alias)
                ->condition($alias . '.' . $record_key, array_keys($values), 'IN')
                ->orderBy($alias . '.' . $record_key);
            $table_mapping = $this->getTableMapping();
            if ($this->revisionDataTable) {
                // Find revisioned fields that are not entity keys. Exclude the langcode
                // key as the base table holds only the default language.
                $base_fields = array_diff($table_mapping->getFieldNames($this->baseTable), [
                    $this->langcodeKey,
                ]);
                $revisioned_fields = array_diff($table_mapping->getFieldNames($this->revisionDataTable), $base_fields);
                // Find fields that are not revisioned or entity keys. Data fields have
                // the same value regardless of entity revision.
                $data_fields = array_diff($table_mapping->getFieldNames($this->dataTable), $revisioned_fields, $base_fields);
                // If there are no data fields then only revisioned fields are needed
                // else both data fields and revisioned fields are needed to map the
                // entity values.
                $all_fields = $revisioned_fields;
                if ($data_fields) {
                    $all_fields = array_merge($revisioned_fields, $data_fields);
                    $query->leftJoin($this->dataTable, 'data', "([revision].[{$this->idKey}] = [data].[{$this->idKey}] AND [revision].[{$this->langcodeKey}] = [data].[{$this->langcodeKey}])");
                    $column_names = [];
                    // Some fields can have more then one columns in the data table so
                    // column names are needed.
                    foreach ($data_fields as $data_field) {
                        // \Drupal\Core\Entity\Sql\TableMappingInterface::getColumnNames()
                        // returns an array keyed by property names so remove the keys
                        // before array_merge() to avoid losing data with fields having the
                        // same columns i.e. value.
                        $column_names = array_merge($column_names, array_values($table_mapping->getColumnNames($data_field)));
                    }
                    $query->fields('data', $column_names);
                }
                // Get the revision IDs.
                $revision_ids = [];
                foreach ($values as $entity_values) {
                    $revision_ids[] = $entity_values[$this->revisionKey][LanguageInterface::LANGCODE_DEFAULT];
                }
                $query->condition('revision.' . $this->revisionKey, $revision_ids, 'IN');
            }
            else {
                $all_fields = $table_mapping->getFieldNames($this->dataTable);
            }
            $result = $query->execute();
            foreach ($result as $row) {
                $id = $row[$record_key];
                // Field values in default language are stored with
                // LanguageInterface::LANGCODE_DEFAULT as key.
                $langcode = empty($row[$this->defaultLangcodeKey]) ? $row[$this->langcodeKey] : LanguageInterface::LANGCODE_DEFAULT;
                $translations[$id][$langcode] = TRUE;
                foreach ($all_fields as $field_name) {
                    $storage_definition = $this->fieldStorageDefinitions[$field_name];
                    $definition_columns = $storage_definition->getColumns();
                    $columns = $table_mapping->getColumnNames($field_name);
                    // Do not key single-column fields by property name.
                    if (count($columns) == 1) {
                        $column_name = reset($columns);
                        $column_attributes = $definition_columns[key($columns)];
                        $values[$id][$field_name][$langcode] = !empty($column_attributes['serialize']) ? unserialize($row[$column_name]) : $row[$column_name];
                    }
                    else {
                        foreach ($columns as $property_name => $column_name) {
                            $column_attributes = $definition_columns[$property_name];
                            $values[$id][$field_name][$langcode][$property_name] = !empty($column_attributes['serialize']) ? unserialize($row[$column_name]) : $row[$column_name];
                        }
                    }
                }
            }
        }
    }
    
    /**
     * {@inheritdoc}
     */
    protected function doLoadMultipleRevisionsFieldItems($revision_ids) {
        $revisions = [];
        // Sanitize IDs. Before feeding ID array into buildQuery, check whether
        // it is empty as this would load all entity revisions.
        $revision_ids = $this->cleanIds($revision_ids, 'revision');
        if (!empty($revision_ids)) {
            // Build and execute the query.
            $query_result = $this->buildQuery(NULL, $revision_ids)
                ->execute();
            $records = $query_result->fetchAllAssoc($this->revisionKey);
            // Map the loaded records into entity objects and according fields.
            if ($records) {
                $revisions = $this->mapFromStorageRecords($records, TRUE);
            }
        }
        return $revisions;
    }
    
    /**
     * {@inheritdoc}
     */
    protected function doDeleteRevisionFieldItems(ContentEntityInterface $revision) {
        $this->database
            ->delete($this->revisionTable)
            ->condition($this->revisionKey, $revision->getRevisionId())
            ->execute();
        if ($this->revisionDataTable) {
            $this->database
                ->delete($this->revisionDataTable)
                ->condition($this->revisionKey, $revision->getRevisionId())
                ->execute();
        }
        $this->deleteRevisionFromDedicatedTables($revision);
    }
    
    /**
     * {@inheritdoc}
     */
    protected function buildPropertyQuery(QueryInterface $entity_query, array $values) {
        if ($this->dataTable) {
            // @todo We should not be using a condition to specify whether conditions
            //   apply to the default language. See
            //   https://www.drupal.org/node/1866330.
            // Default to the original entity language if not explicitly specified
            // otherwise.
            if (!array_key_exists($this->defaultLangcodeKey, $values)) {
                $values[$this->defaultLangcodeKey] = 1;
            }
            elseif ($values[$this->defaultLangcodeKey] === NULL) {
                unset($values[$this->defaultLangcodeKey]);
            }
        }
        parent::buildPropertyQuery($entity_query, $values);
    }
    
    /**
     * Builds the query to load the entity.
     *
     * This has full revision support. For entities requiring special queries,
     * the class can be extended, and the default query can be constructed by
     * calling parent::buildQuery(). This is usually necessary when the object
     * being loaded needs to be augmented with additional data from another
     * table, such as loading vocabulary machine name into terms, however it
     * can also support $conditions on different tables.
     *
     * @param array|null $ids
     *   An array of entity IDs, or NULL to load all entities.
     * @param array|bool $revision_ids
     *   The IDs of the revisions to load, or FALSE if this query is asking for
     *   the default revisions. Defaults to FALSE.
     *
     * @return \Drupal\Core\Database\Query\SelectInterface
     *   A SelectQuery object for loading the entity.
     */
    protected function buildQuery($ids, $revision_ids = FALSE) {
        $query = $this->database
            ->select($this->baseTable, 'base');
        $query->addTag($this->entityTypeId . '_load_multiple');
        if ($revision_ids) {
            $query->join($this->revisionTable, 'revision', "[revision].[{$this->idKey}] = [base].[{$this->idKey}] AND [revision].[{$this->revisionKey}] IN (:revisionIds[])", [
                ':revisionIds[]' => $revision_ids,
            ]);
        }
        elseif ($this->revisionTable) {
            $query->join($this->revisionTable, 'revision', "[revision].[{$this->revisionKey}] = [base].[{$this->revisionKey}]");
        }
        // Add fields from the {entity} table.
        $table_mapping = $this->getTableMapping();
        $entity_fields = $table_mapping->getAllColumns($this->baseTable);
        if ($this->revisionTable) {
            // Add all fields from the {entity_revision} table.
            $entity_revision_fields = $table_mapping->getAllColumns($this->revisionTable);
            $entity_revision_fields = array_combine($entity_revision_fields, $entity_revision_fields);
            // The ID field is provided by entity, so remove it.
            unset($entity_revision_fields[$this->idKey]);
            // Remove all fields from the base table that are also fields by the same
            // name in the revision table.
            $entity_field_keys = array_flip($entity_fields);
            foreach ($entity_revision_fields as $name) {
                if (isset($entity_field_keys[$name])) {
                    unset($entity_fields[$entity_field_keys[$name]]);
                }
            }
            $query->fields('revision', $entity_revision_fields);
            // Compare revision ID of the base and revision table, if equal then this
            // is the default revision.
            $query->addExpression('CASE [base].[' . $this->revisionKey . '] WHEN [revision].[' . $this->revisionKey . '] THEN 1 ELSE 0 END', 'isDefaultRevision');
        }
        $query->fields('base', $entity_fields);
        if ($ids) {
            $query->condition("base.{$this->idKey}", $ids, 'IN');
        }
        return $query;
    }
    
    /**
     * {@inheritdoc}
     */
    public function delete(array $entities) {
        if (!$entities) {
            // If no IDs or invalid IDs were passed, do nothing.
            return;
        }
        try {
            $transaction = $this->database
                ->startTransaction();
            parent::delete($entities);
            // Ignore replica server temporarily.
            \Drupal::service('database.replica_kill_switch')->trigger();
        } catch (\Exception $e) {
            if (isset($transaction)) {
                $transaction->rollBack();
            }
            watchdog_exception($this->entityTypeId, $e);
            throw new EntityStorageException($e->getMessage(), $e->getCode(), $e);
        }
    }
    
    /**
     * {@inheritdoc}
     */
    protected function doDeleteFieldItems($entities) {
        $ids = array_keys($entities);
        $this->database
            ->delete($this->baseTable)
            ->condition($this->idKey, $ids, 'IN')
            ->execute();
        if ($this->revisionTable) {
            $this->database
                ->delete($this->revisionTable)
                ->condition($this->idKey, $ids, 'IN')
                ->execute();
        }
        if ($this->dataTable) {
            $this->database
                ->delete($this->dataTable)
                ->condition($this->idKey, $ids, 'IN')
                ->execute();
        }
        if ($this->revisionDataTable) {
            $this->database
                ->delete($this->revisionDataTable)
                ->condition($this->idKey, $ids, 'IN')
                ->execute();
        }
        foreach ($entities as $entity) {
            $this->deleteFromDedicatedTables($entity);
        }
    }
    
    /**
     * {@inheritdoc}
     */
    public function save(EntityInterface $entity) {
        try {
            $transaction = $this->database
                ->startTransaction();
            $return = parent::save($entity);
            // Ignore replica server temporarily.
            \Drupal::service('database.replica_kill_switch')->trigger();
            return $return;
        } catch (\Exception $e) {
            if (isset($transaction)) {
                $transaction->rollBack();
            }
            watchdog_exception($this->entityTypeId, $e);
            throw new EntityStorageException($e->getMessage(), $e->getCode(), $e);
        }
    }
    
    /**
     * {@inheritdoc}
     */
    public function restore(EntityInterface $entity) {
        try {
            $transaction = $this->database
                ->startTransaction();
            // Insert the entity data in the base and data tables only for default
            // revisions.
            
            /** @var \Drupal\Core\Entity\ContentEntityInterface $entity */
            if ($entity->isDefaultRevision()) {
                $record = $this->mapToStorageRecord($entity->getUntranslated(), $this->baseTable);
                $this->database
                    ->insert($this->baseTable)
                    ->fields((array) $record)
                    ->execute();
                if ($this->dataTable) {
                    $this->saveToSharedTables($entity);
                }
            }
            // Insert the entity data in the revision and revision data tables.
            if ($this->revisionTable) {
                $record = $this->mapToStorageRecord($entity->getUntranslated(), $this->revisionTable);
                $this->database
                    ->insert($this->revisionTable)
                    ->fields((array) $record)
                    ->execute();
                if ($this->revisionDataTable) {
                    $this->saveToSharedTables($entity, $this->revisionDataTable);
                }
            }
            // Insert the entity data in the dedicated tables.
            $this->saveToDedicatedTables($entity, FALSE, []);
            // Ignore replica server temporarily.
            \Drupal::service('database.replica_kill_switch')->trigger();
        } catch (\Exception $e) {
            if (isset($transaction)) {
                $transaction->rollBack();
            }
            watchdog_exception($this->entityTypeId, $e);
            throw new EntityStorageException($e->getMessage(), $e->getCode(), $e);
        }
    }
    
    /**
     * {@inheritdoc}
     */
    protected function doSaveFieldItems(ContentEntityInterface $entity, array $names = []) {
        $full_save = empty($names);
        $update = !$full_save || !$entity->isNew();
        if ($full_save) {
            $shared_table_fields = TRUE;
            $dedicated_table_fields = TRUE;
        }
        else {
            $table_mapping = $this->getTableMapping();
            $shared_table_fields = FALSE;
            $dedicated_table_fields = [];
            // Collect the name of fields to be written in dedicated tables and check
            // whether shared table records need to be updated.
            foreach ($names as $name) {
                $storage_definition = $this->fieldStorageDefinitions[$name];
                if ($table_mapping->allowsSharedTableStorage($storage_definition)) {
                    $shared_table_fields = TRUE;
                }
                elseif ($table_mapping->requiresDedicatedTableStorage($storage_definition)) {
                    $dedicated_table_fields[] = $name;
                }
            }
        }
        // Update shared table records if necessary.
        if ($shared_table_fields) {
            $record = $this->mapToStorageRecord($entity->getUntranslated(), $this->baseTable);
            // Create the storage record to be saved.
            if ($update) {
                $default_revision = $entity->isDefaultRevision();
                if ($default_revision) {
                    $id = $record->{$this->idKey};
                    // Remove the ID from the record to enable updates on SQL variants
                    // that prevent updating serial columns, for example, mssql.
                    unset($record->{$this->idKey});
                    $this->database
                        ->update($this->baseTable)
                        ->fields((array) $record)
                        ->condition($this->idKey, $id)
                        ->execute();
                }
                if ($this->revisionTable) {
                    if ($full_save) {
                        $entity->{$this->revisionKey} = $this->saveRevision($entity);
                    }
                    else {
                        $record = $this->mapToStorageRecord($entity->getUntranslated(), $this->revisionTable);
                        // Remove the revision ID from the record to enable updates on SQL
                        // variants that prevent updating serial columns, for example,
                        // mssql.
                        unset($record->{$this->revisionKey});
                        $entity->preSaveRevision($this, $record);
                        $this->database
                            ->update($this->revisionTable)
                            ->fields((array) $record)
                            ->condition($this->revisionKey, $entity->getRevisionId())
                            ->execute();
                    }
                }
                if ($default_revision && $this->dataTable) {
                    $this->saveToSharedTables($entity);
                }
                if ($this->revisionDataTable) {
                    $new_revision = $full_save && $entity->isNewRevision();
                    $this->saveToSharedTables($entity, $this->revisionDataTable, $new_revision);
                }
            }
            else {
                // @todo Remove the 'return' option in Drupal 11.
                // @see https://www.drupal.org/project/drupal/issues/3256524
                $insert_id = $this->database
                    ->insert($this->baseTable, [
                    'return' => Database::RETURN_INSERT_ID,
                ])
                    ->fields((array) $record)
                    ->execute();
                // Even if this is a new entity the ID key might have been set, in which
                // case we should not override the provided ID. An ID key that is not set
                // to any value is interpreted as NULL (or DEFAULT) and thus overridden.
                if (!isset($record->{$this->idKey})) {
                    $record->{$this->idKey} = $insert_id;
                }
                $entity->{$this->idKey} = (string) $record->{$this->idKey};
                if ($this->revisionTable) {
                    $record->{$this->revisionKey} = $this->saveRevision($entity);
                }
                if ($this->dataTable) {
                    $this->saveToSharedTables($entity);
                }
                if ($this->revisionDataTable) {
                    $this->saveToSharedTables($entity, $this->revisionDataTable);
                }
            }
        }
        // Update dedicated table records if necessary.
        if ($dedicated_table_fields) {
            $names = is_array($dedicated_table_fields) ? $dedicated_table_fields : [];
            $this->saveToDedicatedTables($entity, $update, $names);
        }
    }
    
    /**
     * {@inheritdoc}
     */
    protected function has($id, EntityInterface $entity) {
        return !$entity->isNew();
    }
    
    /**
     * Saves fields that use the shared tables.
     *
     * @param \Drupal\Core\Entity\ContentEntityInterface $entity
     *   The entity object.
     * @param string $table_name
     *   (optional) The table name to save to. Defaults to the data table.
     * @param bool $new_revision
     *   (optional) Whether we are dealing with a new revision. By default fetches
     *   the information from the entity object.
     */
    protected function saveToSharedTables(ContentEntityInterface $entity, $table_name = NULL, $new_revision = NULL) {
        if (!isset($table_name)) {
            $table_name = $this->dataTable;
        }
        if (!isset($new_revision)) {
            $new_revision = $entity->isNewRevision();
        }
        $revision = $table_name != $this->dataTable;
        if (!$revision || !$new_revision) {
            $key = $revision ? $this->revisionKey : $this->idKey;
            $value = $revision ? $entity->getRevisionId() : $entity->id();
            // Delete and insert to handle removed values.
            $this->database
                ->delete($table_name)
                ->condition($key, $value)
                ->execute();
        }
        $query = $this->database
            ->insert($table_name);
        foreach ($entity->getTranslationLanguages() as $langcode => $language) {
            $translation = $entity->getTranslation($langcode);
            $record = $this->mapToDataStorageRecord($translation, $table_name);
            $values = (array) $record;
            $query->fields(array_keys($values))
                ->values($values);
        }
        $query->execute();
    }
    
    /**
     * Maps from an entity object to the storage record.
     *
     * @param \Drupal\Core\Entity\ContentEntityInterface $entity
     *   The entity object.
     * @param string $table_name
     *   (optional) The table name to map records to. Defaults to the base table.
     *
     * @return object
     *   The record to store.
     */
    protected function mapToStorageRecord(ContentEntityInterface $entity, $table_name = NULL) {
        if (!isset($table_name)) {
            $table_name = $this->baseTable;
        }
        $record = new \stdClass();
        $table_mapping = $this->getTableMapping();
        foreach ($table_mapping->getFieldNames($table_name) as $field_name) {
            if (empty($this->fieldStorageDefinitions[$field_name])) {
                throw new EntityStorageException("Table mapping contains invalid field {$field_name}.");
            }
            $definition = $this->fieldStorageDefinitions[$field_name];
            $columns = $table_mapping->getColumnNames($field_name);
            foreach ($columns as $column_name => $schema_name) {
                // If there is no main property and only a single column, get all
                // properties from the first field item and assume that they will be
                // stored serialized.
                // @todo Give field types more control over this behavior in
                //   https://www.drupal.org/node/2232427.
                if (!$definition->getMainPropertyName() && count($columns) == 1) {
                    $value = ($item = $entity->{$field_name}
                        ->first()) ? $item->getValue() : [];
                }
                else {
                    $value = $entity->{$field_name}->{$column_name} ?? NULL;
                }
                if (!empty($definition->getSchema()['columns'][$column_name]['serialize'])) {
                    $value = serialize($value);
                }
                // Do not set serial fields if we do not have a value. This supports all
                // SQL database drivers.
                // @see https://www.drupal.org/node/2279395
                $value = SqlContentEntityStorageSchema::castValue($definition->getSchema()['columns'][$column_name], $value);
                $empty_serial = empty($value) && $this->isColumnSerial($table_name, $schema_name);
                // The user entity is a very special case where the ID field is a serial
                // but we need to insert a row with an ID of 0 to represent the
                // anonymous user.
                // @todo https://drupal.org/i/3222123 implement a generic fix for all
                //   entity types.
                $user_zero = $this->entityTypeId === 'user' && $value === 0;
                if (!$empty_serial || $user_zero) {
                    $record->{$schema_name} = $value;
                }
            }
        }
        return $record;
    }
    
    /**
     * Checks whether a field column should be treated as serial.
     *
     * @param $table_name
     *   The name of the table the field column belongs to.
     * @param $schema_name
     *   The schema name of the field column.
     *
     * @return bool
     *   TRUE if the column is serial, FALSE otherwise.
     */
    protected function isColumnSerial($table_name, $schema_name) {
        $result = FALSE;
        switch ($table_name) {
            case $this->baseTable:
                $result = $schema_name == $this->idKey;
                break;
            case $this->revisionTable:
                $result = $schema_name == $this->revisionKey;
                break;
        }
        return $result;
    }
    
    /**
     * Maps from an entity object to the storage record of the field data.
     *
     * @param \Drupal\Core\Entity\EntityInterface $entity
     *   The entity object.
     * @param string $table_name
     *   (optional) The table name to map records to. Defaults to the data table.
     *
     * @return object
     *   The record to store.
     */
    protected function mapToDataStorageRecord(EntityInterface $entity, $table_name = NULL) {
        if (!isset($table_name)) {
            $table_name = $this->dataTable;
        }
        $record = $this->mapToStorageRecord($entity, $table_name);
        return $record;
    }
    
    /**
     * Saves an entity revision.
     *
     * @param \Drupal\Core\Entity\ContentEntityInterface $entity
     *   The entity object.
     *
     * @return int
     *   The revision id.
     */
    protected function saveRevision(ContentEntityInterface $entity) {
        $record = $this->mapToStorageRecord($entity->getUntranslated(), $this->revisionTable);
        $entity->preSaveRevision($this, $record);
        if ($entity->isNewRevision()) {
            // @todo Remove the 'return' option in Drupal 11.
            // @see https://www.drupal.org/project/drupal/issues/3256524
            $insert_id = $this->database
                ->insert($this->revisionTable, [
                'return' => Database::RETURN_INSERT_ID,
            ])
                ->fields((array) $record)
                ->execute();
            // Even if this is a new revision, the revision ID key might have been
            // set in which case we should not override the provided revision ID.
            if (!isset($record->{$this->revisionKey})) {
                $record->{$this->revisionKey} = $insert_id;
            }
            if ($entity->isDefaultRevision()) {
                $this->database
                    ->update($this->baseTable)
                    ->fields([
                    $this->revisionKey => $record->{$this->revisionKey},
                ])
                    ->condition($this->idKey, $record->{$this->idKey})
                    ->execute();
            }
            // Make sure to update the new revision key for the entity.
            $entity->{$this->revisionKey}->value = $record->{$this->revisionKey};
        }
        else {
            // Remove the revision ID from the record to enable updates on SQL
            // variants that prevent updating serial columns, for example,
            // mssql.
            unset($record->{$this->revisionKey});
            $this->database
                ->update($this->revisionTable)
                ->fields((array) $record)
                ->condition($this->revisionKey, $entity->getRevisionId())
                ->execute();
        }
        return $entity->getRevisionId();
    }
    
    /**
     * {@inheritdoc}
     */
    protected function getQueryServiceName() {
        return 'entity.query.sql';
    }
    
    /**
     * Loads values of fields stored in dedicated tables for a group of entities.
     *
     * @param array &$values
     *   An array of values keyed by entity ID.
     * @param bool $load_from_revision
     *   Flag to indicate whether revisions should be loaded or not.
     */
    protected function loadFromDedicatedTables(array &$values, $load_from_revision) {
        if (empty($values)) {
            return;
        }
        // Collect entities ids, bundles and languages.
        $bundles = [];
        $ids = [];
        $default_langcodes = [];
        foreach ($values as $key => $entity_values) {
            $bundles[$this->bundleKey ? $entity_values[$this->bundleKey][LanguageInterface::LANGCODE_DEFAULT] : $this->entityTypeId] = TRUE;
            $ids[] = !$load_from_revision ? $key : $entity_values[$this->revisionKey][LanguageInterface::LANGCODE_DEFAULT];
            if ($this->langcodeKey && isset($entity_values[$this->langcodeKey][LanguageInterface::LANGCODE_DEFAULT])) {
                $default_langcodes[$key] = $entity_values[$this->langcodeKey][LanguageInterface::LANGCODE_DEFAULT];
            }
        }
        // Collect impacted fields.
        $storage_definitions = [];
        $definitions = [];
        $table_mapping = $this->getTableMapping();
        foreach ($bundles as $bundle => $v) {
            $definitions[$bundle] = $this->entityFieldManager
                ->getFieldDefinitions($this->entityTypeId, $bundle);
            foreach ($definitions[$bundle] as $field_name => $field_definition) {
                $storage_definition = $field_definition->getFieldStorageDefinition();
                if ($table_mapping->requiresDedicatedTableStorage($storage_definition)) {
                    $storage_definitions[$field_name] = $storage_definition;
                }
            }
        }
        // Load field data.
        $langcodes = array_keys($this->languageManager
            ->getLanguages(LanguageInterface::STATE_ALL));
        foreach ($storage_definitions as $field_name => $storage_definition) {
            $table = !$load_from_revision ? $table_mapping->getDedicatedDataTableName($storage_definition) : $table_mapping->getDedicatedRevisionTableName($storage_definition);
            // Ensure that only values having valid languages are retrieved. Since we
            // are loading values for multiple entities, we cannot limit the query to
            // the available translations.
            $results = $this->database
                ->select($table, 't')
                ->fields('t')
                ->condition(!$load_from_revision ? 'entity_id' : 'revision_id', $ids, 'IN')
                ->condition('deleted', 0)
                ->condition('langcode', $langcodes, 'IN')
                ->orderBy('delta')
                ->execute();
            foreach ($results as $row) {
                $bundle = $row->bundle;
                $value_key = !$load_from_revision ? $row->entity_id : $row->revision_id;
                // Field values in default language are stored with
                // LanguageInterface::LANGCODE_DEFAULT as key.
                $langcode = LanguageInterface::LANGCODE_DEFAULT;
                if ($this->langcodeKey && isset($default_langcodes[$value_key]) && $row->langcode != $default_langcodes[$value_key]) {
                    $langcode = $row->langcode;
                }
                if (!isset($values[$value_key][$field_name][$langcode])) {
                    $values[$value_key][$field_name][$langcode] = [];
                }
                // Ensure that records for non-translatable fields having invalid
                // languages are skipped.
                if ($langcode == LanguageInterface::LANGCODE_DEFAULT || $definitions[$bundle][$field_name]->isTranslatable()) {
                    if ($storage_definition->getCardinality() == FieldStorageDefinitionInterface::CARDINALITY_UNLIMITED || count($values[$value_key][$field_name][$langcode]) < $storage_definition->getCardinality()) {
                        $item = [];
                        // For each column declared by the field, populate the item from the
                        // prefixed database column.
                        foreach ($storage_definition->getColumns() as $column => $attributes) {
                            $column_name = $table_mapping->getFieldColumnName($storage_definition, $column);
                            // Unserialize the value if specified in the column schema.
                            $item[$column] = !empty($attributes['serialize']) ? unserialize($row->{$column_name}) : $row->{$column_name};
                        }
                        // Add the item to the field values for the entity.
                        $values[$value_key][$field_name][$langcode][] = $item;
                    }
                }
            }
        }
    }
    
    /**
     * Saves values of fields that use dedicated tables.
     *
     * @param \Drupal\Core\Entity\ContentEntityInterface $entity
     *   The entity.
     * @param bool $update
     *   TRUE if the entity is being updated, FALSE if it is being inserted.
     * @param string[] $names
     *   (optional) The names of the fields to be stored. Defaults to all the
     *   available fields.
     */
    protected function saveToDedicatedTables(ContentEntityInterface $entity, $update = TRUE, $names = []) {
        $vid = $entity->getRevisionId();
        $id = $entity->id();
        $bundle = $entity->bundle();
        $entity_type = $entity->getEntityTypeId();
        $default_langcode = $entity->getUntranslated()
            ->language()
            ->getId();
        $translation_langcodes = array_keys($entity->getTranslationLanguages());
        $table_mapping = $this->getTableMapping();
        if (!isset($vid)) {
            $vid = $id;
        }
        $original = !empty($entity->original) ? $entity->original : NULL;
        // Determine which fields should be actually stored.
        $definitions = $this->entityFieldManager
            ->getFieldDefinitions($entity_type, $bundle);
        if ($names) {
            $definitions = array_intersect_key($definitions, array_flip($names));
        }
        foreach ($definitions as $field_name => $field_definition) {
            $storage_definition = $field_definition->getFieldStorageDefinition();
            if (!$table_mapping->requiresDedicatedTableStorage($storage_definition)) {
                continue;
            }
            // When updating an existing revision, keep the existing records if the
            // field values did not change.
            if (!$entity->isNewRevision() && $original && !$this->hasFieldValueChanged($field_definition, $entity, $original)) {
                continue;
            }
            $table_name = $table_mapping->getDedicatedDataTableName($storage_definition);
            $revision_name = $table_mapping->getDedicatedRevisionTableName($storage_definition);
            // Delete and insert, rather than update, in case a value was added.
            if ($update) {
                // Only overwrite the field's base table if saving the default revision
                // of an entity.
                if ($entity->isDefaultRevision()) {
                    $this->database
                        ->delete($table_name)
                        ->condition('entity_id', $id)
                        ->execute();
                }
                if ($this->entityType
                    ->isRevisionable()) {
                    $this->database
                        ->delete($revision_name)
                        ->condition('entity_id', $id)
                        ->condition('revision_id', $vid)
                        ->execute();
                }
            }
            // Prepare the multi-insert query.
            $do_insert = FALSE;
            $columns = [
                'entity_id',
                'revision_id',
                'bundle',
                'delta',
                'langcode',
            ];
            foreach ($storage_definition->getColumns() as $column => $attributes) {
                $columns[] = $table_mapping->getFieldColumnName($storage_definition, $column);
            }
            $query = $this->database
                ->insert($table_name)
                ->fields($columns);
            if ($this->entityType
                ->isRevisionable()) {
                $revision_query = $this->database
                    ->insert($revision_name)
                    ->fields($columns);
            }
            $langcodes = $field_definition->isTranslatable() ? $translation_langcodes : [
                $default_langcode,
            ];
            foreach ($langcodes as $langcode) {
                $delta_count = 0;
                $items = $entity->getTranslation($langcode)
                    ->get($field_name);
                $items->filterEmptyItems();
                foreach ($items as $delta => $item) {
                    // We now know we have something to insert.
                    $do_insert = TRUE;
                    $record = [
                        'entity_id' => $id,
                        'revision_id' => $vid,
                        'bundle' => $bundle,
                        'delta' => $delta,
                        'langcode' => $langcode,
                    ];
                    foreach ($storage_definition->getColumns() as $column => $attributes) {
                        $column_name = $table_mapping->getFieldColumnName($storage_definition, $column);
                        // Serialize the value if specified in the column schema.
                        $value = $item->{$column};
                        if (!empty($attributes['serialize'])) {
                            $value = serialize($value);
                        }
                        $record[$column_name] = SqlContentEntityStorageSchema::castValue($attributes, $value);
                    }
                    $query->values($record);
                    if ($this->entityType
                        ->isRevisionable()) {
                        $revision_query->values($record);
                    }
                    if ($storage_definition->getCardinality() != FieldStorageDefinitionInterface::CARDINALITY_UNLIMITED && ++$delta_count == $storage_definition->getCardinality()) {
                        break;
                    }
                }
            }
            // Execute the query if we have values to insert.
            if ($do_insert) {
                // Only overwrite the field's base table if saving the default revision
                // of an entity.
                if ($entity->isDefaultRevision()) {
                    $query->execute();
                }
                if ($this->entityType
                    ->isRevisionable()) {
                    $revision_query->execute();
                }
            }
        }
    }
    
    /**
     * Deletes values of fields in dedicated tables for all revisions.
     *
     * @param \Drupal\Core\Entity\ContentEntityInterface $entity
     *   The entity.
     */
    protected function deleteFromDedicatedTables(ContentEntityInterface $entity) {
        $table_mapping = $this->getTableMapping();
        foreach ($this->fieldStorageDefinitions as $storage_definition) {
            if (!$table_mapping->requiresDedicatedTableStorage($storage_definition)) {
                continue;
            }
            $table_name = $table_mapping->getDedicatedDataTableName($storage_definition);
            $revision_name = $table_mapping->getDedicatedRevisionTableName($storage_definition);
            $this->database
                ->delete($table_name)
                ->condition('entity_id', $entity->id())
                ->execute();
            if ($this->entityType
                ->isRevisionable()) {
                $this->database
                    ->delete($revision_name)
                    ->condition('entity_id', $entity->id())
                    ->execute();
            }
        }
    }
    
    /**
     * Deletes values of fields in dedicated tables for all revisions.
     *
     * @param \Drupal\Core\Entity\ContentEntityInterface $entity
     *   The entity. It must have a revision ID.
     */
    protected function deleteRevisionFromDedicatedTables(ContentEntityInterface $entity) {
        $vid = $entity->getRevisionId();
        if (isset($vid)) {
            $table_mapping = $this->getTableMapping();
            foreach ($this->fieldStorageDefinitions as $storage_definition) {
                if (!$table_mapping->requiresDedicatedTableStorage($storage_definition)) {
                    continue;
                }
                $revision_name = $table_mapping->getDedicatedRevisionTableName($storage_definition);
                $this->database
                    ->delete($revision_name)
                    ->condition('entity_id', $entity->id())
                    ->condition('revision_id', $vid)
                    ->execute();
            }
        }
    }
    
    /**
     * {@inheritdoc}
     */
    public function requiresEntityStorageSchemaChanges(EntityTypeInterface $entity_type, EntityTypeInterface $original) {
        return $this->getStorageSchema()
            ->requiresEntityStorageSchemaChanges($entity_type, $original);
    }
    
    /**
     * {@inheritdoc}
     */
    public function requiresFieldStorageSchemaChanges(FieldStorageDefinitionInterface $storage_definition, FieldStorageDefinitionInterface $original) {
        return $this->getStorageSchema()
            ->requiresFieldStorageSchemaChanges($storage_definition, $original);
    }
    
    /**
     * {@inheritdoc}
     */
    public function requiresEntityDataMigration(EntityTypeInterface $entity_type, EntityTypeInterface $original) {
        return $this->getStorageSchema()
            ->requiresEntityDataMigration($entity_type, $original);
    }
    
    /**
     * {@inheritdoc}
     */
    public function requiresFieldDataMigration(FieldStorageDefinitionInterface $storage_definition, FieldStorageDefinitionInterface $original) {
        return $this->getStorageSchema()
            ->requiresFieldDataMigration($storage_definition, $original);
    }
    
    /**
     * {@inheritdoc}
     */
    public function onEntityTypeCreate(EntityTypeInterface $entity_type) {
        $this->wrapSchemaException(function () use ($entity_type) {
            $this->getStorageSchema()
                ->onEntityTypeCreate($entity_type);
        });
    }
    
    /**
     * {@inheritdoc}
     */
    public function onEntityTypeUpdate(EntityTypeInterface $entity_type, EntityTypeInterface $original) {
        // Ensure we have an updated entity type definition.
        $this->entityType = $entity_type;
        // The table layout may have changed depending on the new entity type
        // definition.
        $this->initTableLayout();
        // Let the schema handler adapt to possible table layout changes.
        $this->wrapSchemaException(function () use ($entity_type, $original) {
            $this->getStorageSchema()
                ->onEntityTypeUpdate($entity_type, $original);
        });
    }
    
    /**
     * {@inheritdoc}
     */
    public function onEntityTypeDelete(EntityTypeInterface $entity_type) {
        $this->wrapSchemaException(function () use ($entity_type) {
            $this->getStorageSchema()
                ->onEntityTypeDelete($entity_type);
        });
    }
    
    /**
     * {@inheritdoc}
     */
    public function onFieldableEntityTypeCreate(EntityTypeInterface $entity_type, array $field_storage_definitions) {
        $this->wrapSchemaException(function () use ($entity_type, $field_storage_definitions) {
            $this->getStorageSchema()
                ->onFieldableEntityTypeCreate($entity_type, $field_storage_definitions);
        });
    }
    
    /**
     * {@inheritdoc}
     */
    public function onFieldableEntityTypeUpdate(EntityTypeInterface $entity_type, EntityTypeInterface $original, array $field_storage_definitions, array $original_field_storage_definitions, array &$sandbox = NULL) {
        $this->wrapSchemaException(function () use ($entity_type, $original, $field_storage_definitions, $original_field_storage_definitions, &$sandbox) {
            $this->getStorageSchema()
                ->onFieldableEntityTypeUpdate($entity_type, $original, $field_storage_definitions, $original_field_storage_definitions, $sandbox);
        });
    }
    
    /**
     * {@inheritdoc}
     */
    public function onFieldStorageDefinitionCreate(FieldStorageDefinitionInterface $storage_definition) {
        $this->wrapSchemaException(function () use ($storage_definition) {
            $this->getStorageSchema()
                ->onFieldStorageDefinitionCreate($storage_definition);
            $this->fieldStorageDefinitions[$storage_definition->getName()] = $storage_definition;
            $this->tableMapping = NULL;
        });
    }
    
    /**
     * {@inheritdoc}
     */
    public function onFieldStorageDefinitionUpdate(FieldStorageDefinitionInterface $storage_definition, FieldStorageDefinitionInterface $original) {
        $this->wrapSchemaException(function () use ($storage_definition, $original) {
            $this->getStorageSchema()
                ->onFieldStorageDefinitionUpdate($storage_definition, $original);
            $this->fieldStorageDefinitions[$storage_definition->getName()] = $storage_definition;
            $this->tableMapping = NULL;
        });
    }
    
    /**
     * {@inheritdoc}
     */
    public function onFieldStorageDefinitionDelete(FieldStorageDefinitionInterface $storage_definition) {
        $table_mapping = $this->getTableMapping();
        if ($table_mapping->requiresDedicatedTableStorage($storage_definition)) {
            // Mark all data associated with the field for deletion.
            $table = $table_mapping->getDedicatedDataTableName($storage_definition);
            $revision_table = $table_mapping->getDedicatedRevisionTableName($storage_definition);
            $this->database
                ->update($table)
                ->fields([
                'deleted' => 1,
            ])
                ->execute();
            if ($this->entityType
                ->isRevisionable()) {
                $this->database
                    ->update($revision_table)
                    ->fields([
                    'deleted' => 1,
                ])
                    ->execute();
            }
        }
        // Update the field schema.
        $this->wrapSchemaException(function () use ($storage_definition) {
            $this->getStorageSchema()
                ->onFieldStorageDefinitionDelete($storage_definition);
            unset($this->fieldStorageDefinitions[$storage_definition->getName()]);
            $this->tableMapping = NULL;
        });
    }
    
    /**
     * Wraps a database schema exception into an entity storage exception.
     *
     * @param callable $callback
     *   The callback to be executed.
     *
     * @throws \Drupal\Core\Entity\EntityStorageException
     *   When a database schema exception is thrown.
     */
    protected function wrapSchemaException(callable $callback) {
        $message = 'Exception thrown while performing a schema update.';
        try {
            $callback();
        } catch (SchemaException $e) {
            $message .= ' ' . $e->getMessage();
            throw new EntityStorageException($message, 0, $e);
        } catch (DatabaseExceptionWrapper $e) {
            $message .= ' ' . $e->getMessage();
            throw new EntityStorageException($message, 0, $e);
        }
    }
    
    /**
     * {@inheritdoc}
     */
    public function onFieldDefinitionDelete(FieldDefinitionInterface $field_definition) {
        $table_mapping = $this->getTableMapping();
        $storage_definition = $field_definition->getFieldStorageDefinition();
        // Mark field data as deleted.
        if ($table_mapping->requiresDedicatedTableStorage($storage_definition)) {
            $table_name = $table_mapping->getDedicatedDataTableName($storage_definition);
            $revision_name = $table_mapping->getDedicatedRevisionTableName($storage_definition);
            $this->database
                ->update($table_name)
                ->fields([
                'deleted' => 1,
            ])
                ->condition('bundle', $field_definition->getTargetBundle())
                ->execute();
            if ($this->entityType
                ->isRevisionable()) {
                $this->database
                    ->update($revision_name)
                    ->fields([
                    'deleted' => 1,
                ])
                    ->condition('bundle', $field_definition->getTargetBundle())
                    ->execute();
            }
        }
    }
    
    /**
     * {@inheritdoc}
     */
    public function onBundleCreate($bundle, $entity_type_id) {
    }
    
    /**
     * {@inheritdoc}
     */
    public function onBundleDelete($bundle, $entity_type_id) {
    }
    
    /**
     * {@inheritdoc}
     */
    protected function readFieldItemsToPurge(FieldDefinitionInterface $field_definition, $batch_size) {
        // Check whether the whole field storage definition is gone, or just some
        // bundle fields.
        $storage_definition = $field_definition->getFieldStorageDefinition();
        $table_mapping = $this->getTableMapping();
        $table_name = $table_mapping->getDedicatedDataTableName($storage_definition, $storage_definition->isDeleted());
        // Get the entities which we want to purge first.
        $entity_query = $this->database
            ->select($table_name, 't', [
            'fetch' => \PDO::FETCH_ASSOC,
        ]);
        $or = $entity_query->orConditionGroup();
        foreach ($storage_definition->getColumns() as $column_name => $data) {
            $or->isNotNull($table_mapping->getFieldColumnName($storage_definition, $column_name));
        }
        $entity_query->distinct(TRUE)
            ->fields('t', [
            'entity_id',
        ])
            ->condition('bundle', $field_definition->getTargetBundle())
            ->range(0, $batch_size);
        // Create a map of field data table column names to field column names.
        $column_map = [];
        foreach ($storage_definition->getColumns() as $column_name => $data) {
            $column_map[$table_mapping->getFieldColumnName($storage_definition, $column_name)] = $column_name;
        }
        $entities = [];
        $items_by_entity = [];
        foreach ($entity_query->execute() as $row) {
            $item_query = $this->database
                ->select($table_name, 't', [
                'fetch' => \PDO::FETCH_ASSOC,
            ])
                ->fields('t')
                ->condition('entity_id', $row['entity_id'])
                ->condition('deleted', 1)
                ->orderBy('delta');
            foreach ($item_query->execute() as $item_row) {
                if (!isset($entities[$item_row['revision_id']])) {
                    // Create entity with the right revision id and entity id combination.
                    $item_row['entity_type'] = $this->entityTypeId;
                    // @todo: Replace this by an entity object created via an entity
                    // factory, see https://www.drupal.org/node/1867228.
                    $entities[$item_row['revision_id']] = _field_create_entity_from_ids((object) $item_row);
                }
                $item = [];
                foreach ($column_map as $db_column => $field_column) {
                    $item[$field_column] = $item_row[$db_column];
                }
                $items_by_entity[$item_row['revision_id']][] = $item;
            }
        }
        // Create field item objects and return.
        foreach ($items_by_entity as $revision_id => $values) {
            $entity_adapter = $entities[$revision_id]->getTypedData();
            $items_by_entity[$revision_id] = \Drupal::typedDataManager()->create($field_definition, $values, $field_definition->getName(), $entity_adapter);
        }
        return $items_by_entity;
    }
    
    /**
     * {@inheritdoc}
     */
    protected function purgeFieldItems(ContentEntityInterface $entity, FieldDefinitionInterface $field_definition) {
        $storage_definition = $field_definition->getFieldStorageDefinition();
        $is_deleted = $storage_definition->isDeleted();
        $table_mapping = $this->getTableMapping();
        $table_name = $table_mapping->getDedicatedDataTableName($storage_definition, $is_deleted);
        $revision_name = $table_mapping->getDedicatedRevisionTableName($storage_definition, $is_deleted);
        $revision_id = $this->entityType
            ->isRevisionable() ? $entity->getRevisionId() : $entity->id();
        $this->database
            ->delete($table_name)
            ->condition('revision_id', $revision_id)
            ->condition('deleted', 1)
            ->execute();
        if ($this->entityType
            ->isRevisionable()) {
            $this->database
                ->delete($revision_name)
                ->condition('revision_id', $revision_id)
                ->condition('deleted', 1)
                ->execute();
        }
    }
    
    /**
     * {@inheritdoc}
     */
    public function finalizePurge(FieldStorageDefinitionInterface $storage_definition) {
        $this->getStorageSchema()
            ->finalizePurge($storage_definition);
    }
    
    /**
     * {@inheritdoc}
     */
    public function countFieldData($storage_definition, $as_bool = FALSE) {
        // Ensure that the table mapping is instantiated with the passed-in field
        // storage definition.
        $storage_definitions = $this->fieldStorageDefinitions;
        $storage_definitions[$storage_definition->getName()] = $storage_definition;
        $table_mapping = $this->getTableMapping($storage_definitions);
        if ($table_mapping->requiresDedicatedTableStorage($storage_definition)) {
            $is_deleted = $storage_definition->isDeleted();
            if ($this->entityType
                ->isRevisionable()) {
                $table_name = $table_mapping->getDedicatedRevisionTableName($storage_definition, $is_deleted);
            }
            else {
                $table_name = $table_mapping->getDedicatedDataTableName($storage_definition, $is_deleted);
            }
            $query = $this->database
                ->select($table_name, 't');
            $or = $query->orConditionGroup();
            foreach ($storage_definition->getColumns() as $column_name => $data) {
                $or->isNotNull($table_mapping->getFieldColumnName($storage_definition, $column_name));
            }
            $query->condition($or);
            if (!$as_bool) {
                $query->fields('t', [
                    'entity_id',
                ])
                    ->distinct(TRUE);
            }
        }
        elseif ($table_mapping->allowsSharedTableStorage($storage_definition)) {
            // Ascertain the table this field is mapped too.
            $field_name = $storage_definition->getName();
            $table_name = $table_mapping->getFieldTableName($field_name);
            $query = $this->database
                ->select($table_name, 't');
            $or = $query->orConditionGroup();
            foreach (array_keys($storage_definition->getColumns()) as $property_name) {
                $or->isNotNull($table_mapping->getFieldColumnName($storage_definition, $property_name));
            }
            $query->condition($or);
            if (!$as_bool) {
                $query->fields('t', [
                    $this->idKey,
                ])
                    ->distinct(TRUE);
            }
        }
        // @todo Find a way to count field data also for fields having custom
        //   storage. See https://www.drupal.org/node/2337753.
        $count = 0;
        if (isset($query)) {
            // If we are performing the query just to check if the field has data
            // limit the number of rows.
            if ($as_bool) {
                $query->range(0, 1)
                    ->addExpression('1');
            }
            else {
                // Otherwise count the number of rows.
                $query = $query->countQuery();
            }
            $count = $query->execute()
                ->fetchField();
        }
        return $as_bool ? (bool) $count : (int) $count;
    }

}