diff --git a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java index 59d9ff75ae3..ee9e1ea7f88 100644 --- a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java +++ b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java @@ -36,7 +36,7 @@ public class CassandraSchemaVersionManager { public static final SchemaVersion MIN_VERSION = new SchemaVersion(12); - public static final SchemaVersion MAX_VERSION = new SchemaVersion(15); + public static final SchemaVersion MAX_VERSION = new SchemaVersion(16); public static final SchemaVersion DEFAULT_VERSION = MIN_VERSION; private static final Logger LOGGER = LoggerFactory.getLogger(CassandraSchemaVersionManager.class); diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java index 375d61bc93e..49701316485 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java @@ -32,7 +32,6 @@ import static org.apache.james.mailbox.cassandra.table.CassandraMessageIds.MESSAGE_ID; import static org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.ATTACHMENTS; import static org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.BODY_CONTENT; -import static org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.BODY_OCTECTS; import static org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.BODY_START_OCTET; import static org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.FULL_CONTENT_OCTETS; import static org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.HEADER_CONTENT; @@ -158,7 +157,6 @@ private PreparedStatement prepareInsert(CqlSession session) { .set(setColumn(INTERNAL_DATE, bindMarker(INTERNAL_DATE)), setColumn(BODY_START_OCTET, bindMarker(BODY_START_OCTET)), setColumn(FULL_CONTENT_OCTETS, bindMarker(FULL_CONTENT_OCTETS)), - setColumn(BODY_OCTECTS, bindMarker(BODY_OCTECTS)), setColumn(BODY_CONTENT, bindMarker(BODY_CONTENT)), setColumn(HEADER_CONTENT, bindMarker(HEADER_CONTENT)), prepend(ATTACHMENTS, bindMarker(ATTACHMENTS))) @@ -184,7 +182,6 @@ public Mono save(MessageRepresentation message) { .setInstant(INTERNAL_DATE, message.getInternalDate().toInstant()) .setInt(BODY_START_OCTET, message.getBodyStartOctet()) .setLong(FULL_CONTENT_OCTETS, message.getSize()) - .setLong(BODY_OCTECTS, message.getSize() - message.getBodyStartOctet()) .setString(BODY_CONTENT, message.getBodyId().asString()) .setString(HEADER_CONTENT, message.getHeaderId().asString()); @@ -227,7 +224,6 @@ private BoundStatement boundWriteStatement(MailboxMessage message, Tuple2 initializer) { private final BlobId.Factory blobIdFactory; private final PreparedStatement delete; private final PreparedStatement insert; + private final PreparedStatement updateDenormalizedFields; private final PreparedStatement select; private final PreparedStatement selectAll; private final PreparedStatement selectAllUids; @@ -144,6 +144,7 @@ public CassandraMessageIdDAO(CqlSession session, BlobId.Factory blobIdFactory) { this.delete = prepareDelete(session); this.insert = prepareInsert(session); this.update = prepareUpdate(session); + this.updateDenormalizedFields = prepareUpdateDenormalizedFields(session); this.select = prepareSelect(session); this.selectAll = prepareSelectAll(session); this.selectAllUids = prepareSelectAllUids(session); @@ -179,7 +180,6 @@ private PreparedStatement prepareInsert(CqlSession session) { setColumn(FLAGGED, bindMarker(FLAGGED)), setColumn(RECENT, bindMarker(RECENT)), setColumn(SEEN, bindMarker(SEEN)), - setColumn(USER, bindMarker(USER)), setColumn(INTERNAL_DATE, bindMarker(INTERNAL_DATE)), setColumn(SAVE_DATE, bindMarker(SAVE_DATE)), setColumn(BODY_START_OCTET, bindMarker(BODY_START_OCTET)), @@ -191,6 +191,17 @@ private PreparedStatement prepareInsert(CqlSession session) { .build()); } + private PreparedStatement prepareUpdateDenormalizedFields(CqlSession session) { + return session.prepare(update(TABLE_NAME) + .set(setColumn(INTERNAL_DATE, bindMarker(INTERNAL_DATE)), + setColumn(BODY_START_OCTET, bindMarker(BODY_START_OCTET)), + setColumn(FULL_CONTENT_OCTETS, bindMarker(FULL_CONTENT_OCTETS)), + setColumn(HEADER_CONTENT, bindMarker(HEADER_CONTENT))) + .where(column(MAILBOX_ID).isEqualTo(bindMarker(MAILBOX_ID)), + column(IMAP_UID).isEqualTo(bindMarker(IMAP_UID))) + .build()); + } + private PreparedStatement prepareUpdate(CqlSession session) { return session.prepare(update(TABLE_NAME) .set(setColumn(MOD_SEQ, bindMarker(MOD_SEQ)), @@ -200,7 +211,6 @@ private PreparedStatement prepareUpdate(CqlSession session) { setColumn(FLAGGED, bindMarker(FLAGGED)), setColumn(RECENT, bindMarker(RECENT)), setColumn(SEEN, bindMarker(SEEN)), - setColumn(USER, bindMarker(USER)), append(USER_FLAGS, bindMarker(ADDED_USERS_FLAGS)), remove(USER_FLAGS, bindMarker(REMOVED_USERS_FLAGS))) .where(column(MAILBOX_ID).isEqualTo(bindMarker(MAILBOX_ID)), @@ -297,7 +307,6 @@ private PreparedStatement prepareSelectMetadataRange(CqlSession session) { RECENT, SEEN, FLAGGED, - USER, USER_FLAGS, MOD_SEQ) .where(column(MAILBOX_ID).isEqualTo(bindMarker(MAILBOX_ID)), @@ -359,7 +368,6 @@ public Mono insert(CassandraMessageMetadata metadata) { .setBoolean(FLAGGED, flags.contains(Flag.FLAGGED)) .setBoolean(RECENT, flags.contains(Flag.RECENT)) .setBoolean(SEEN, flags.contains(Flag.SEEN)) - .setBoolean(USER, flags.contains(Flag.USER)) .setInstant(INTERNAL_DATE, metadata.getInternalDate().get().toInstant()) .setInstant(SAVE_DATE, metadata.getSaveDate().map(Date::toInstant).orElse(null)) .setInt(BODY_START_OCTET, Math.toIntExact(metadata.getBodyStartOctet().get())) @@ -369,6 +377,17 @@ public Mono insert(CassandraMessageMetadata metadata) { .build()); } + public Mono updateDenormalizedFields(CassandraId mailboxId, MessageUid uid, Date internalDate, + int bodyStartOctet, long size, BlobId headerContent) { + return cassandraAsyncExecutor.executeVoid(updateDenormalizedFields.bind() + .setUuid(MAILBOX_ID, mailboxId.asUuid()) + .setLong(IMAP_UID, uid.asLong()) + .setInstant(INTERNAL_DATE, internalDate.toInstant()) + .setInt(BODY_START_OCTET, bodyStartOctet) + .setLong(FULL_CONTENT_OCTETS, size) + .setString(HEADER_CONTENT, headerContent.asString())); + } + public Mono updateMetadata(ComposedMessageId composedMessageId, UpdatedFlags updatedFlags) { return cassandraAsyncExecutor.executeVoid(updateBoundStatement(composedMessageId, updatedFlags)); } @@ -409,11 +428,6 @@ private BoundStatement updateBoundStatement(ComposedMessageId id, UpdatedFlags u } else { statementBuilder.unset(SEEN); } - if (updatedFlags.isChanged(Flag.USER)) { - statementBuilder.setBoolean(USER, updatedFlags.isModifiedToSet(Flag.USER)); - } else { - statementBuilder.unset(USER); - } Sets.SetView removedFlags = Sets.difference( ImmutableSet.copyOf(updatedFlags.getOldFlags().getUserFlags()), ImmutableSet.copyOf(updatedFlags.getNewFlags().getUserFlags())); @@ -647,7 +661,6 @@ Mono insertNullInternalDateAndHeaderContent(CassandraMessageMetadata metad .setBoolean(FLAGGED, flags.contains(Flag.FLAGGED)) .setBoolean(RECENT, flags.contains(Flag.RECENT)) .setBoolean(SEEN, flags.contains(Flag.SEEN)) - .setBoolean(USER, flags.contains(Flag.USER)) .setInstant(INTERNAL_DATE, null) .setInt(BODY_START_OCTET, 0) .setLong(FULL_CONTENT_OCTETS, 0) diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAO.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAO.java index 6f793a0cef6..a57973e133f 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAO.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAO.java @@ -40,11 +40,11 @@ import static org.apache.james.mailbox.cassandra.table.Flag.FLAGGED; import static org.apache.james.mailbox.cassandra.table.Flag.RECENT; import static org.apache.james.mailbox.cassandra.table.Flag.SEEN; -import static org.apache.james.mailbox.cassandra.table.Flag.USER; import static org.apache.james.mailbox.cassandra.table.Flag.USER_FLAGS; import static org.apache.james.mailbox.cassandra.table.MessageIdToImapUid.MOD_SEQ; import static org.apache.james.mailbox.cassandra.table.MessageIdToImapUid.TABLE_NAME; import static org.apache.james.mailbox.cassandra.table.MessageIdToImapUid.THREAD_ID; +import static org.apache.james.util.ReactorUtils.publishIfPresent; import java.time.Duration; import java.util.Date; @@ -87,6 +87,7 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; public class CassandraMessageIdToImapUidDAO { private static final String MOD_SEQ_CONDITION = "modSeqCondition"; @@ -98,6 +99,7 @@ public class CassandraMessageIdToImapUidDAO { private final PreparedStatement delete; private final PreparedStatement insert; private final PreparedStatement update; + private final PreparedStatement updateDenormalizedFields; private final PreparedStatement selectAll; private final PreparedStatement select; private final PreparedStatement listStatement; @@ -116,6 +118,7 @@ public CassandraMessageIdToImapUidDAO(CqlSession session, BlobId.Factory blobIdF this.cassandraConfiguration = cassandraConfiguration; this.delete = prepareDelete(); this.insert = prepareInsert(); + this.updateDenormalizedFields = prepareUpdateDenormalizedFields(); this.update = prepareUpdate(); this.selectAll = prepareSelectAll(); this.select = prepareSelect(); @@ -145,7 +148,6 @@ private PreparedStatement prepareInsert() { .value(FLAGGED, bindMarker(FLAGGED)) .value(RECENT, bindMarker(RECENT)) .value(SEEN, bindMarker(SEEN)) - .value(USER, bindMarker(USER)) .value(USER_FLAGS, bindMarker(USER_FLAGS)) .value(INTERNAL_DATE, bindMarker(INTERNAL_DATE)) .value(SAVE_DATE, bindMarker(SAVE_DATE)) @@ -164,7 +166,6 @@ private PreparedStatement prepareInsert() { setColumn(FLAGGED, bindMarker(FLAGGED)), setColumn(RECENT, bindMarker(RECENT)), setColumn(SEEN, bindMarker(SEEN)), - setColumn(USER, bindMarker(USER)), setColumn(INTERNAL_DATE, bindMarker(INTERNAL_DATE)), setColumn(SAVE_DATE, bindMarker(SAVE_DATE)), setColumn(BODY_START_OCTET, bindMarker(BODY_START_OCTET)), @@ -178,6 +179,18 @@ private PreparedStatement prepareInsert() { } } + private PreparedStatement prepareUpdateDenormalizedFields() { + return session.prepare(QueryBuilder.update(TABLE_NAME) + .set(setColumn(INTERNAL_DATE, bindMarker(INTERNAL_DATE)), + setColumn(BODY_START_OCTET, bindMarker(BODY_START_OCTET)), + setColumn(FULL_CONTENT_OCTETS, bindMarker(FULL_CONTENT_OCTETS)), + setColumn(HEADER_CONTENT, bindMarker(HEADER_CONTENT))) + .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)), + column(MAILBOX_ID).isEqualTo(bindMarker(MAILBOX_ID)), + column(IMAP_UID).isEqualTo(bindMarker(IMAP_UID))) + .build()); + } + private PreparedStatement prepareUpdate() { Update update = QueryBuilder.update(TABLE_NAME) .set(setColumn(MOD_SEQ, bindMarker(MOD_SEQ)), @@ -186,8 +199,7 @@ private PreparedStatement prepareUpdate() { setColumn(DRAFT, bindMarker(DRAFT)), setColumn(FLAGGED, bindMarker(FLAGGED)), setColumn(RECENT, bindMarker(RECENT)), - setColumn(SEEN, bindMarker(SEEN)), - setColumn(USER, bindMarker(USER))) + setColumn(SEEN, bindMarker(SEEN))) .append(USER_FLAGS, bindMarker(ADDED_USERS_FLAGS)) .remove(USER_FLAGS, bindMarker(REMOVED_USERS_FLAGS)) .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)), @@ -251,7 +263,6 @@ public Mono insert(CassandraMessageMetadata metadata) { .setBoolean(FLAGGED, flags.contains(Flag.FLAGGED)) .setBoolean(RECENT, flags.contains(Flag.RECENT)) .setBoolean(SEEN, flags.contains(Flag.SEEN)) - .setBoolean(USER, flags.contains(Flag.USER)) .setInstant(INTERNAL_DATE, metadata.getInternalDate().get().toInstant()) .setInstant(SAVE_DATE, metadata.getSaveDate().map(Date::toInstant).orElse(null)) .setInt(BODY_START_OCTET, Math.toIntExact(metadata.getBodyStartOctet().get())) @@ -261,6 +272,18 @@ public Mono insert(CassandraMessageMetadata metadata) { .build()); } + public Mono updateDenormalizedFields(CassandraMessageId messageId, CassandraId mailboxId, MessageUid uid, + Date internalDate, int bodyStartOctet, long size, BlobId headerContent) { + return cassandraAsyncExecutor.executeVoid(updateDenormalizedFields.bind() + .setUuid(MESSAGE_ID, messageId.get()) + .setUuid(MAILBOX_ID, mailboxId.asUuid()) + .setLong(IMAP_UID, uid.asLong()) + .setInstant(INTERNAL_DATE, internalDate.toInstant()) + .setInt(BODY_START_OCTET, bodyStartOctet) + .setLong(FULL_CONTENT_OCTETS, size) + .setString(HEADER_CONTENT, headerContent.asString())); + } + public Mono updateMetadata(ComposedMessageId id, UpdatedFlags updatedFlags, ModSeq previousModeq) { if (cassandraConfiguration.isMessageWriteStrongConsistency()) { return cassandraAsyncExecutor.executeReturnApplied(updateBoundStatement(id, updatedFlags, previousModeq)); @@ -306,11 +329,6 @@ private BoundStatement updateBoundStatement(ComposedMessageId id, UpdatedFlags u } else { statementBuilder.unset(SEEN); } - if (updatedFlags.isChanged(Flag.USER)) { - statementBuilder.setBoolean(USER, updatedFlags.isModifiedToSet(Flag.USER)); - } else { - statementBuilder.unset(USER); - } Sets.SetView removedFlags = Sets.difference( ImmutableSet.copyOf(updatedFlags.getOldFlags().getUserFlags()), ImmutableSet.copyOf(updatedFlags.getNewFlags().getUserFlags())); @@ -335,7 +353,8 @@ private BoundStatement updateBoundStatement(ComposedMessageId id, UpdatedFlags u public Flux retrieve(CassandraMessageId messageId, Optional mailboxId, JamesExecutionProfiles.ConsistencyChoice readConsistencyChoice) { return cassandraAsyncExecutor.executeRows(setExecutionProfileIfNeeded(selectStatement(messageId, mailboxId), readConsistencyChoice)) - .map(this::toComposedMessageIdWithMetadata); + .map(this::toComposedMessageIdWithMetadata) + .handle(publishIfPresent()); } @VisibleForTesting @@ -346,12 +365,22 @@ public Flux retrieve(CassandraMessageId messageId, Opt public Flux retrieveAllMessages() { return cassandraAsyncExecutor.executeRows(listStatement.bind() .setTimeout(Duration.ofDays(1))) - .map(this::toComposedMessageIdWithMetadata); + .map(this::toComposedMessageIdWithMetadata) + .handle(publishIfPresent()); } - private CassandraMessageMetadata toComposedMessageIdWithMetadata(Row row) { + private Optional toComposedMessageIdWithMetadata(Row row) { final CassandraMessageId messageId = CassandraMessageId.Factory.of(row.getUuid(MESSAGE_ID)); - return CassandraMessageMetadata.builder() + if (row.get(MOD_SEQ, Long.class) == null) { + // Out of order updates with concurrent deletes can result in the row being partially deleted + // We filter out such records, and cleanup them. + // TODO Test INTERNAL_DATE instead once schema version 16 is enforced: unlike MOD_SEQ it also catches rows resurrected by a flag update. + delete(messageId, CassandraId.of(row.getUuid(MAILBOX_ID))) + .subscribeOn(Schedulers.parallel()) + .subscribe(); + return Optional.empty(); + } + return Optional.of(CassandraMessageMetadata.builder() .ids(ComposedMessageIdWithMetaData.builder() .composedMessageId(new ComposedMessageId( CassandraId.of(row.getUuid(MAILBOX_ID)), @@ -369,7 +398,7 @@ private CassandraMessageMetadata toComposedMessageIdWithMetadata(Row row) { .size(row.get(FULL_CONTENT_OCTETS, Long.class)) .headerContent(Optional.ofNullable(row.getString(HEADER_CONTENT)) .map(blobIdFactory::parse)) - .build(); + .build()); } private ThreadId getThreadIdFromRow(Row row, MessageId messageId) { @@ -421,7 +450,6 @@ Mono insertNullInternalDateAndHeaderContent(CassandraMessageMetadata metad .setBoolean(FLAGGED, flags.contains(Flag.FLAGGED)) .setBoolean(RECENT, flags.contains(Flag.RECENT)) .setBoolean(SEEN, flags.contains(Flag.SEEN)) - .setBoolean(USER, flags.contains(Flag.USER)) .setInstant(INTERNAL_DATE, null) .setInt(BODY_START_OCTET, 0) .setLong(FULL_CONTENT_OCTETS, 0) diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/MessageDenormalizationMigration.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/MessageDenormalizationMigration.java new file mode 100644 index 00000000000..26d9c5d0643 --- /dev/null +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/MessageDenormalizationMigration.java @@ -0,0 +1,130 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ + +package org.apache.james.mailbox.cassandra.mail.migration; + +import java.util.function.Function; + +import jakarta.inject.Inject; + +import org.apache.james.backends.cassandra.migration.Migration; +import org.apache.james.mailbox.cassandra.ids.CassandraId; +import org.apache.james.mailbox.cassandra.ids.CassandraMessageId; +import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAOV3; +import org.apache.james.mailbox.cassandra.mail.CassandraMessageIdDAO; +import org.apache.james.mailbox.cassandra.mail.CassandraMessageIdToImapUidDAO; +import org.apache.james.mailbox.cassandra.mail.CassandraMessageMetadata; +import org.apache.james.mailbox.cassandra.mail.MessageRepresentation; +import org.apache.james.mailbox.model.ComposedMessageId; +import org.apache.james.mailbox.store.mail.MessageMapper.FetchType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * Backfills the fields `messageIdTable` and `imapUidTable` denormalize from `messagev3`. + * + *

Those columns were introduced by JAMES-3576 in 3.7.0 and never backfilled: JAMES-3815 chose instead + * to tolerate their absence, which is what {@link CassandraMessageMetadata#isComplete()} tests, falling + * back to `messagev3` when they are missing. Messages written by James 3.6 and earlier therefore still + * carry null there.

+ * + *

Once every row is complete, that fallback becomes dead: metadata and header fetches are answered + * from a single read, and `messagev3` no longer needs to carry the denormalized copies at all.

+ * + * See JAMES-4225 + */ +public class MessageDenormalizationMigration implements Migration { + private static final Logger LOGGER = LoggerFactory.getLogger(MessageDenormalizationMigration.class); + private static final int CONCURRENCY = 8; + + private final CassandraMessageIdDAO messageIdDAO; + private final CassandraMessageIdToImapUidDAO imapUidDAO; + private final CassandraMessageDAOV3 messageDAO; + + @Inject + public MessageDenormalizationMigration(CassandraMessageIdDAO messageIdDAO, + CassandraMessageIdToImapUidDAO imapUidDAO, + CassandraMessageDAOV3 messageDAO) { + this.messageIdDAO = messageIdDAO; + this.imapUidDAO = imapUidDAO; + this.messageDAO = messageDAO; + } + + @Override + public void apply() { + backfill() + .then(cleanUpPartialRows()) + .block(); + } + + private Mono backfill() { + return Flux.concat( + backfill(imapUidDAO.retrieveAllMessages(), this::backfillImapUid), + backfill(messageIdDAO.retrieveAllMessages(), this::backfillMessageId)) + .then(); + } + + private Mono cleanUpPartialRows() { + return Flux.concat(imapUidDAO.retrieveAllMessages(), messageIdDAO.retrieveAllMessages()) + .then(); + } + + private Flux backfill(Flux rows, + Function> backfill) { + return rows.filter(metadata -> !metadata.isComplete()) + .flatMap(backfill, CONCURRENCY); + } + + private Mono backfillMessageId(CassandraMessageMetadata metadata) { + ComposedMessageId id = metadata.getComposedMessageId().getComposedMessageId(); + + return representation(metadata) + .flatMap(representation -> messageIdDAO.updateDenormalizedFields( + (CassandraId) id.getMailboxId(), + id.getUid(), + representation.getInternalDate(), + representation.getBodyStartOctet(), + representation.getSize(), + representation.getHeaderId())); + } + + private Mono backfillImapUid(CassandraMessageMetadata metadata) { + ComposedMessageId id = metadata.getComposedMessageId().getComposedMessageId(); + + return representation(metadata) + .flatMap(representation -> imapUidDAO.updateDenormalizedFields( + (CassandraMessageId) id.getMessageId(), + (CassandraId) id.getMailboxId(), + id.getUid(), + representation.getInternalDate(), + representation.getBodyStartOctet(), + representation.getSize(), + representation.getHeaderId())); + } + + private Mono representation(CassandraMessageMetadata metadata) { + return messageDAO.retrieveMessage(metadata.getComposedMessageId(), FetchType.METADATA) + .doOnError(e -> LOGGER.error("Failed to read messagev3 for {}", + metadata.getComposedMessageId().getComposedMessageId(), e)) + .onErrorResume(e -> Mono.empty()); + } +} diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/modules/CassandraMessageDataDefinition.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/modules/CassandraMessageDataDefinition.java index e5b87ff04a3..92169be3557 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/modules/CassandraMessageDataDefinition.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/modules/CassandraMessageDataDefinition.java @@ -61,7 +61,6 @@ public interface CassandraMessageDataDefinition { .withColumn(Flag.FLAGGED, BOOLEAN) .withColumn(Flag.RECENT, BOOLEAN) .withColumn(Flag.SEEN, BOOLEAN) - .withColumn(Flag.USER, BOOLEAN) .withColumn(Flag.USER_FLAGS, setOf(TEXT)) .withColumn(CassandraMessageV3Table.INTERNAL_DATE, TIMESTAMP) .withColumn(CassandraMessageIdTable.SAVE_DATE, TIMESTAMP) @@ -86,7 +85,6 @@ public interface CassandraMessageDataDefinition { .withColumn(Flag.FLAGGED, BOOLEAN) .withColumn(Flag.RECENT, BOOLEAN) .withColumn(Flag.SEEN, BOOLEAN) - .withColumn(Flag.USER, BOOLEAN) .withColumn(Flag.USER_FLAGS, setOf(TEXT)) .withColumn(CassandraMessageV3Table.INTERNAL_DATE, TIMESTAMP) .withColumn(CassandraMessageIdTable.SAVE_DATE, TIMESTAMP) @@ -100,7 +98,6 @@ public interface CassandraMessageDataDefinition { .withPartitionKey(CassandraMessageIds.MESSAGE_ID, TIMEUUID) .withColumn(CassandraMessageV3Table.INTERNAL_DATE, TIMESTAMP) .withColumn(CassandraMessageV3Table.BODY_START_OCTET, INT) - .withColumn(CassandraMessageV3Table.BODY_OCTECTS, BIGINT) .withColumn(CassandraMessageV3Table.FULL_CONTENT_OCTETS, BIGINT) .withColumn(CassandraMessageV3Table.BODY_CONTENT, TEXT) .withColumn(CassandraMessageV3Table.HEADER_CONTENT, TEXT) diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/CassandraMessageV3Table.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/CassandraMessageV3Table.java index d78445fc082..975be163602 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/CassandraMessageV3Table.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/CassandraMessageV3Table.java @@ -27,7 +27,6 @@ public interface CassandraMessageV3Table { CqlIdentifier INTERNAL_DATE = CqlIdentifier.fromCql("internalDate"); CqlIdentifier BODY_START_OCTET = CqlIdentifier.fromCql("bodyStartOctet"); CqlIdentifier FULL_CONTENT_OCTETS = CqlIdentifier.fromCql("fullContentOctets"); - CqlIdentifier BODY_OCTECTS = CqlIdentifier.fromCql("bodyOctets"); CqlIdentifier BODY_CONTENT = CqlIdentifier.fromCql("bodyContent"); CqlIdentifier HEADER_CONTENT = CqlIdentifier.fromCql("headerContent"); CqlIdentifier ATTACHMENTS = CqlIdentifier.fromCql("attachments"); diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/Flag.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/Flag.java index fdf367b5ac9..ac6a5ac011d 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/Flag.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/Flag.java @@ -30,7 +30,6 @@ public interface Flag { CqlIdentifier RECENT = CqlIdentifier.fromCql("flagRecent"); CqlIdentifier SEEN = CqlIdentifier.fromCql("flagSeen"); CqlIdentifier FLAGGED = CqlIdentifier.fromCql("flagFlagged"); - CqlIdentifier USER = CqlIdentifier.fromCql("flagUser"); CqlIdentifier USER_FLAGS = CqlIdentifier.fromCql("userFlags"); CqlIdentifier[] ALL_LOWERCASE = { @@ -39,8 +38,7 @@ public interface Flag { DRAFT, RECENT, SEEN, - FLAGGED, - USER + FLAGGED }; ImmutableMap JAVAX_MAIL_FLAG = ImmutableMap.builder() @@ -50,6 +48,5 @@ public interface Flag { .put(RECENT, Flags.Flag.RECENT) .put(SEEN, Flags.Flag.SEEN) .put(FLAGGED, Flags.Flag.FLAGGED) - .put(USER, Flags.Flag.USER) .build(); } \ No newline at end of file diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdDAOTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdDAOTest.java index 3bb94fd39f2..6aa9370f411 100644 --- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdDAOTest.java +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdDAOTest.java @@ -554,7 +554,7 @@ void updateShouldUpdateSeenFlag() { } @Test - void updateShouldUpdateUserFlag() { + void updateShouldNotPersistUserFlag() { CassandraMessageId messageId = messageIdFactory.generate(); CassandraId mailboxId = CassandraId.timeBased(); MessageUid messageUid = MessageUid.of(1); @@ -574,9 +574,11 @@ void updateShouldUpdateUserFlag() { .build()) .block(); + // \* is IMAP PERMANENTFLAGS syntax, a mailbox capability rather than a flag of a message: no + // protocol lets a client set it on one, so it is not stored. ComposedMessageIdWithMetaData expectedComposedMessageId = ComposedMessageIdWithMetaData.builder() .composedMessageId(composedMessageId) - .flags(new Flags(Flags.Flag.USER)) + .flags(new Flags()) .modSeq(ModSeq.of(2)) .threadId(ThreadId.fromBaseMessageId(messageId)) .build(); @@ -585,7 +587,7 @@ void updateShouldUpdateUserFlag() { .messageId(messageId) .modSeq(expectedComposedMessageId.getModSeq()) .oldFlags(new Flags()) - .newFlags(expectedComposedMessageId.getFlags()) + .newFlags(new Flags(Flags.Flag.USER)) .build(); testee.updateMetadata(expectedComposedMessageId.getComposedMessageId(), updatedFlags) diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAOTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAOTest.java index c2cb662d7a2..b420bc8c16d 100644 --- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAOTest.java +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAOTest.java @@ -604,7 +604,7 @@ void updateShouldUpdateSeenFlag() { } @Test - void updateShouldUpdateUserFlag() { + void updateShouldNotPersistUserFlag() { CassandraMessageId messageId = CassandraMessageId.Factory.of(Uuids.timeBased()); CassandraId mailboxId = CassandraId.timeBased(); MessageUid messageUid = MessageUid.of(1); @@ -624,9 +624,11 @@ void updateShouldUpdateUserFlag() { .build()) .block(); + // \* is IMAP PERMANENTFLAGS syntax, a mailbox capability rather than a flag of a message: no + // protocol lets a client set it on one, so it is not stored. ComposedMessageIdWithMetaData expectedComposedMessageId = ComposedMessageIdWithMetaData.builder() .composedMessageId(composedMessageId) - .flags(new Flags(Flags.Flag.USER)) + .flags(new Flags()) .modSeq(ModSeq.of(2)) .threadId(ThreadId.fromBaseMessageId(messageId)) .build(); @@ -635,7 +637,7 @@ void updateShouldUpdateUserFlag() { .messageId(messageId) .modSeq(ModSeq.of(2)) .oldFlags(new Flags()) - .newFlags(expectedComposedMessageId.getFlags()) + .newFlags(new Flags(Flags.Flag.USER)) .build(); testee.updateMetadata(expectedComposedMessageId.getComposedMessageId(), updatedFlags, ModSeq.of(1)).block(); diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/MessageDenormalizationMigrationTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/MessageDenormalizationMigrationTest.java new file mode 100644 index 00000000000..1aa100736db --- /dev/null +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/MessageDenormalizationMigrationTest.java @@ -0,0 +1,247 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ + +package org.apache.james.mailbox.cassandra.mail; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.nio.charset.StandardCharsets; +import java.util.Date; +import java.util.Optional; + +import jakarta.mail.Flags; + +import org.apache.james.backends.cassandra.CassandraCluster; +import org.apache.james.backends.cassandra.CassandraClusterExtension; +import org.apache.james.backends.cassandra.components.CassandraDataDefinition; +import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration; +import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionDataDefinition; +import org.apache.james.blob.api.BlobStore; +import org.apache.james.blob.api.BlobStoreCacheCallback; +import org.apache.james.blob.api.BucketName; +import org.apache.james.blob.api.PlainBlobId; +import org.apache.james.blob.cassandra.CassandraBlobDataDefinition; +import org.apache.james.blob.cassandra.CassandraBlobStoreDAO; +import org.apache.james.blob.cassandra.CassandraBucketDAO; +import org.apache.james.blob.cassandra.CassandraDefaultBucketDAO; +import org.apache.james.mailbox.MessageUid; +import org.apache.james.mailbox.ModSeq; +import org.apache.james.mailbox.cassandra.ids.CassandraId; +import org.apache.james.mailbox.cassandra.ids.CassandraMessageId; +import org.apache.james.mailbox.cassandra.mail.migration.MessageDenormalizationMigration; +import org.apache.james.mailbox.cassandra.modules.CassandraMessageDataDefinition; +import org.apache.james.mailbox.model.ByteContent; +import org.apache.james.mailbox.model.ComposedMessageId; +import org.apache.james.mailbox.model.ComposedMessageIdWithMetaData; +import org.apache.james.mailbox.model.ThreadId; +import org.apache.james.mailbox.store.mail.model.impl.SimpleMailboxMessage; +import org.apache.james.metrics.tests.RecordingMetricFactory; +import org.apache.james.server.blob.deduplication.BlobStoreFactory; +import org.awaitility.Awaitility; +import org.awaitility.Durations; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import com.google.common.collect.ImmutableList; + +class MessageDenormalizationMigrationTest { + private static final CassandraId MAILBOX_ID = CassandraId.timeBased(); + private static final MessageUid MESSAGE_UID = MessageUid.of(1); + private static final String CONTENT = "Subject: test\n\nBody\n"; + private static final int BODY_START = 15; + + public static final CassandraDataDefinition MODULES = CassandraDataDefinition.aggregateModules( + CassandraMessageDataDefinition.MODULE, + CassandraBlobDataDefinition.MODULE, + CassandraSchemaVersionDataDefinition.MODULE); + + @RegisterExtension + static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(MODULES); + + private CassandraMessageIdDAO messageIdDAO; + private CassandraMessageIdToImapUidDAO imapUidDAO; + private CassandraMessageDAOV3 messageDAO; + private MessageDenormalizationMigration testee; + + private CassandraMessageId messageId; + private ComposedMessageIdWithMetaData ids; + + @BeforeEach + void setUp(CassandraCluster cassandra) { + PlainBlobId.Factory blobIdFactory = new PlainBlobId.Factory(); + CassandraBlobStoreDAO blobStoreDAO = new CassandraBlobStoreDAO( + new CassandraDefaultBucketDAO(cassandra.getConf(), blobIdFactory), + new CassandraBucketDAO(blobIdFactory, cassandra.getConf()), + CassandraConfiguration.DEFAULT_CONFIGURATION, BucketName.DEFAULT, new RecordingMetricFactory()); + BlobStore blobStore = BlobStoreFactory.builder() + .blobStoreDAO(blobStoreDAO) + .blobIdFactory(blobIdFactory) + .defaultBucketName() + .passthrough(); + + messageIdDAO = new CassandraMessageIdDAO(cassandra.getConf(), blobIdFactory); + imapUidDAO = new CassandraMessageIdToImapUidDAO(cassandra.getConf(), blobIdFactory, + CassandraConfiguration.DEFAULT_CONFIGURATION); + messageDAO = new CassandraMessageDAOV3(cassandra.getConf(), cassandra.getTypesProvider(), blobStore, + blobStoreDAO, blobIdFactory, CassandraConfiguration.DEFAULT_CONFIGURATION, BlobStoreCacheCallback.NOOP); + testee = new MessageDenormalizationMigration(messageIdDAO, imapUidDAO, messageDAO); + + messageId = new CassandraMessageId.Factory().generate(); + ids = ComposedMessageIdWithMetaData.builder() + .composedMessageId(new ComposedMessageId(MAILBOX_ID, messageId, MESSAGE_UID)) + .flags(new Flags()) + .modSeq(ModSeq.of(1)) + .threadId(ThreadId.fromBaseMessageId(messageId)) + .build(); + } + + @Test + void migrationShouldCompleteMessageIdRows() throws Exception { + saveMessage(); + givenIncompleteRows(); + + testee.apply(); + + assertThat(retrieveMessageId()).hasValueSatisfying(metadata -> + assertThat(metadata.isComplete()).isTrue()); + } + + @Test + void migrationShouldCompleteImapUidRows() throws Exception { + saveMessage(); + givenIncompleteRows(); + + testee.apply(); + + assertThat(retrieveImapUid()).hasValueSatisfying(metadata -> + assertThat(metadata.isComplete()).isTrue()); + } + + @Test + void migrationShouldCopyTheFieldsOfMessageV3() throws Exception { + SimpleMailboxMessage message = saveMessage(); + givenIncompleteRows(); + + testee.apply(); + + CassandraMessageMetadata metadata = retrieveMessageId().get(); + assertThat(metadata.getInternalDate()).contains(message.getInternalDate()); + assertThat(metadata.getBodyStartOctet()).contains((long) BODY_START); + assertThat(metadata.getSize()).contains((long) CONTENT.length()); + assertThat(metadata.getHeaderContent()).isNotEmpty(); + } + + @Test + void migrationShouldNotAlterCompleteRows() throws Exception { + saveMessage(); + givenCompleteRows(); + CassandraMessageMetadata before = retrieveMessageId().get(); + + testee.apply(); + + assertThat(retrieveMessageId()).contains(before); + } + + @Test + void migrationShouldIgnoreRowsWhoseMessageIsMissing() throws Exception { + givenIncompleteRows(); + + testee.apply(); + + assertThat(retrieveMessageId()).hasValueSatisfying(metadata -> + assertThat(metadata.isComplete()).isFalse()); + } + + /** + * Backfilling is an upsert: a message deleted between the read and the write comes back as a row + * carrying the denormalized columns alone. Writing one directly is the deterministic way to assert + * such a row never surfaces, the race itself not being reproducible. + */ + @Test + void resurrectedMessageIdRowsShouldNotSurface() { + messageIdDAO.updateDenormalizedFields(MAILBOX_ID, MESSAGE_UID, new Date(), BODY_START, + CONTENT.length(), new PlainBlobId.Factory().of("headerBlobId")).block(); + + assertThat(retrieveMessageId()).isEmpty(); + } + + @Test + void resurrectedImapUidRowsShouldNotSurface() { + imapUidDAO.updateDenormalizedFields(messageId, MAILBOX_ID, MESSAGE_UID, new Date(), BODY_START, + CONTENT.length(), new PlainBlobId.Factory().of("headerBlobId")).block(); + + assertThat(retrieveImapUid()).isEmpty(); + } + + @Test + void resurrectedRowsShouldBeCleanedUp() { + messageIdDAO.updateDenormalizedFields(MAILBOX_ID, MESSAGE_UID, new Date(), BODY_START, + CONTENT.length(), new PlainBlobId.Factory().of("headerBlobId")).block(); + + retrieveMessageId(); + + Awaitility.await().atMost(Durations.TEN_SECONDS) + .untilAsserted(() -> assertThat(messageIdDAO.retrieveAllMessages().collectList().block()).isEmpty()); + } + + private SimpleMailboxMessage saveMessage() { + SimpleMailboxMessage message = SimpleMailboxMessage.builder() + .messageId(messageId) + .threadId(ThreadId.fromBaseMessageId(messageId)) + .mailboxId(MAILBOX_ID) + .uid(MESSAGE_UID) + .internalDate(new Date()) + .bodyStartOctet(BODY_START) + .size(CONTENT.length()) + .content(new ByteContent(CONTENT.getBytes(StandardCharsets.UTF_8))) + .flags(new Flags()) + .addAttachments(ImmutableList.of()) + .build(); + messageDAO.save(message).block(); + return message; + } + + private void givenIncompleteRows() { + CassandraMessageMetadata metadata = CassandraMessageMetadata.builder().ids(ids).build(); + messageIdDAO.insertNullInternalDateAndHeaderContent(metadata).block(); + imapUidDAO.insertNullInternalDateAndHeaderContent(metadata).block(); + } + + private void givenCompleteRows() { + CassandraMessageMetadata metadata = CassandraMessageMetadata.builder() + .ids(ids) + .internalDate(new Date()) + .bodyStartOctet((long) BODY_START) + .size((long) CONTENT.length()) + .headerContent(Optional.of(new PlainBlobId.Factory().of("headerBlobId"))) + .build(); + messageIdDAO.insert(metadata).block(); + imapUidDAO.insert(metadata).block(); + } + + private Optional retrieveMessageId() { + return messageIdDAO.retrieve(MAILBOX_ID, MESSAGE_UID).block(); + } + + private Optional retrieveImapUid() { + return imapUidDAO.retrieve(messageId, Optional.of(MAILBOX_ID)).collectList().block() + .stream().findFirst(); + } +} diff --git a/server/container/guice/cassandra/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java b/server/container/guice/cassandra/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java index b4c17d7caca..07d5e79cfde 100644 --- a/server/container/guice/cassandra/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java +++ b/server/container/guice/cassandra/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java @@ -25,6 +25,7 @@ import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionManager; import org.apache.james.backends.cassandra.versions.SchemaTransition; import org.apache.james.backends.cassandra.versions.SchemaVersion; +import org.apache.james.mailbox.cassandra.mail.migration.MessageDenormalizationMigration; import org.apache.james.mailbox.cassandra.quota.migration.CassandraCurrentQuotaManagerMigration; import org.apache.james.mailbox.cassandra.quota.migration.CassandraPerUserMaxQuotaManagerMigration; import org.apache.james.sieve.cassandra.migration.SieveQuotaMigration; @@ -42,6 +43,7 @@ public class CassandraRoutesModule extends AbstractModule { private static final SchemaTransition FROM_V12_TO_V13 = SchemaTransition.to(new SchemaVersion(13)); private static final SchemaTransition FROM_V13_TO_V14 = SchemaTransition.to(new SchemaVersion(14)); private static final SchemaTransition FROM_V14_TO_V15 = SchemaTransition.to(new SchemaVersion(15)); + private static final SchemaTransition FROM_V15_TO_V16 = SchemaTransition.to(new SchemaVersion(16)); @Override protected void configure() { @@ -60,6 +62,7 @@ protected void configure() { allMigrationClazzBinder.addBinding(FROM_V12_TO_V13).to(CassandraCurrentQuotaManagerMigration.class); allMigrationClazzBinder.addBinding(FROM_V13_TO_V14).to(CassandraPerUserMaxQuotaManagerMigration.class); allMigrationClazzBinder.addBinding(FROM_V14_TO_V15).to(SieveQuotaMigration.class); + allMigrationClazzBinder.addBinding(FROM_V15_TO_V16).to(MessageDenormalizationMigration.class); bind(SchemaVersion.class) .annotatedWith(Names.named(CassandraMigrationService.LATEST_VERSION)) diff --git a/upgrade-instructions.md b/upgrade-instructions.md index 2dd12243b57..4e84aba1781 100644 --- a/upgrade-instructions.md +++ b/upgrade-instructions.md @@ -23,6 +23,52 @@ Change list: - [JAMES-4210 POP3 USER/PASS requires TLS by default](#james-4210-pop3-userpass-requires-tls-by-default) - [JAMES-4210 ManageSieve SASL adoption](#james-4210-managesieve-sasl-adoption) - [JAMES-4225 Blob ids default to 128 bits of entropy](#james-4225-blob-ids-default-to-128-bits-of-entropy) + - [Dropping unneeded Cassandra schema columns](#dropping-unneeded-cassandra-schema-columns) + - [Cassandra schema version 16: mandatory message denormalization migration](#cassandra-schema-version-16-mandatory-message-denormalization-migration) + +### Dropping unneeded Cassandra schema columns + +Date: 05/09/2026 + +Concerned products: James products using Cassandra as mailbox storage + +Three columns held no information and are no longer written: + +- `messagev3.bodyOctets`, never read back, the body size being derived from `fullContentOctets` and +`bodyStartOctet`. +- `messageIdTable.flagUser` and `imapUidTable.flagUser`, persisting the JavaMail `Flags.Flag.USER`. That +flag is IMAP `PERMANENTFLAGS` syntax, `\*`, a property of a mailbox rather than of a message, that no +protocol lets a client set on one: the column only ever held `false`. + +New messages stop paying for them right away. Reclaiming the space taken by existing rows requires +dropping the columns manually, once the new version runs on every node: + +```sql +ALTER TABLE james_keyspace.messagev3 DROP bodyOctets; +ALTER TABLE james_keyspace.messageIdTable DROP flagUser; +ALTER TABLE james_keyspace.imapUidTable DROP flagUser; +``` + +Disk space comes back progressively, as SSTables get compacted. + +### Cassandra schema version 16: mandatory message denormalization migration + +Date: 05/09/2026 + +Concerned products: James products using Cassandra as mailbox storage + +`messageIdTable` and `imapUidTable` denormalize four fields from `messagev3`: `internalDate`, +`bodyStartOctet`, `fullContentOctets` and `headerContent`. They were introduced in 3.7.0 and never +backfilled: James tolerates their absence instead, and falls back to reading `messagev3`. Messages +written by James 3.6 and earlier therefore still carry null there. + +Schema version 16 backfills them. Run it as any other migration, for instance: + +``` +curl -XPOST 'http://ip:port/cassandra/version/upgrade' -d '16' +``` + +Later releases will drop this backward support and drop soon-to-be-useless collumns. ### JAMES-4225 Blob ids default to 128 bits of entropy