Repository: duckdb/duckdb
File: src/storage/storage_manager.cpp (C++)

Code before:
			storage_version_user_provided = entry.second.ToString();
			storage_version =
			    SerializationCompatibility::FromString(storage_version_user_provided).serialization_version;
		} else if (entry.first == "compress") {
			if (entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue<bool>()) {
				compress_in_memory = CompressInMemory::COMPRESS;
			} else {
				compress_in_memory = CompressInMemory::DO_NOT_COMPRESS;
			}
		} else if (entry.first == "debug_encryption_version") {
			encryption_version = EncryptionTypes::StringToVersion(entry.second.ToString());
		} else {
			throw BinderException("Unrecognized option for attach \"%s\"", entry.first);
		}
	}
	if (encryption) {
		if (!storage_version.IsValid() ||
		    storage_version.GetIndex() < SerializationCompatibility::FromString("v1.4.0").serialization_version) {
			if (!storage_version_user_provided.empty()) {
				throw InvalidInputException("Explicit provided STORAGE_VERSION (\"%s\") and ENCRYPTION_KEY (storage >= "
				                            "v1.4.0) are not compatible",
				                            storage_version_user_provided);
			}
		}
		// explicit storage version input
		if ((!storage_version_user_provided.empty() &&
		     storage_version.GetIndex() < SerializationCompatibility::FromString("v1.5.0").serialization_version)) {
			// user input and encryption version input v1
			if (encryption_version == EncryptionTypes::V0_1) {
				throw InvalidConfigurationException(
				    "Encryption version explicitly set to V1, but storage version is not compatible");
			}
			// keep storage version, but force encryption version to lower for compatibility
			encryption_version = EncryptionTypes::V0_0;
		}

		if (storage_version_user_provided.empty()) {
			// set storage version explicitly
			storage_version = SerializationCompatibility::FromString("v1.5.0").serialization_version;
		}

		if (encryption_version == EncryptionTypes::NONE) {
			// set default to encryption version v1
			encryption_version = EncryptionTypes::V0_1;
		}
	}
}

StorageManager::StorageManager(AttachedDatabase &db, string path_p, const AttachOptions &options)
    : db(db), path(std::move(path_p)), read_only(options.access_mode == AccessMode::READ_ONLY), wal_size(0) {
	if (path.empty()) {

Review comment (bug):
I still think the following block is not completely correct, there are a few more cases to consider, like I would set storage_version after checking it's not already higher than necessary (currently it will just override v1.5.0, whatever it is to start with).

Note that this might not be wrong now, but it will become in a few months.

Code after:
					throw InvalidInputException("Explicit provided STORAGE_VERSION (\"%s\") is not compatible with "
					                            "'debug_encryption_version = v1' (storage >= "
					                            "v1.5.0)",
					                            storage_version_user_provided);
				}
			} else {
				// encryption version needs to be lowered, because storage version < v1.5.0
				target_encryption_version = EncryptionTypes::V0_0;
				break;
			}
		}

		break;

	case EncryptionTypes::V0_0:
		// we set this to V0 to V1.5.0 if no explicit storage version provided
		if (!storage_version.IsValid() && storage_version_user_provided.empty()) {
			storage_version = SerializationCompatibility::FromString("v1.5.0").serialization_version;
			break;
		}
		// if storage version is provided, we do nothing
		break;
	default:
		throw InvalidConfigurationException("Encryption version is not set");
	}

	encryption_version = target_encryption_version;
}

void StorageOptions::Initialize(const unordered_map<string, Value> &options) {
	string storage_version_user_provided = "";
	for (auto &entry : options) {
		if (entry.first == "block_size") {
			// Extract the block allocation size. This is NOT the actual memory available on a block (block_size),
			// even though the corresponding option we expose to the user is called "block_size".
			block_alloc_size = entry.second.GetValue<uint64_t>();
		} else if (entry.first == "encryption_key") {
			// check the type of the key
			auto type = entry.second.type();
			if (type.id() != LogicalTypeId::VARCHAR) {
				throw BinderException("\"%s\" is not a valid key. A key must be of type VARCHAR",
				                      entry.second.ToString());
			} else if (entry.second.GetValue<string>().empty()) {
				throw BinderException("Not a valid key. A key cannot be empty");
			}
			user_key = make_shared_ptr<string>(StringValue::Get(entry.second.DefaultCastAs(LogicalType::BLOB)));
			block_header_size = DEFAULT_ENCRYPTION_BLOCK_HEADER_SIZE;
			encryption = true;
		} else if (entry.first == "encryption_cipher") {
			auto parsed_cipher = EncryptionTypes::StringToCipher(entry.second.ToString());
			if (parsed_cipher != EncryptionTypes::CipherType::GCM &&