Add a way to configure log spill type.

fdbclient/DatabaseConfiguration.cpp

@@ -31,6 +31,7 @@ void DatabaseConfiguration::resetInternal() {
 	initialized = false;
 	masterProxyCount = resolverCount = desiredTLogCount = tLogWriteAntiQuorum = tLogReplicationFactor = storageTeamSize = desiredLogRouterCount = -1;
 	tLogDataStoreType = storageServerStoreType = KeyValueStoreType::END;
+	tLogSpillType = TLogSpillType::DEFAULT;
 	autoMasterProxyCount = CLIENT_KNOBS->DEFAULT_AUTO_PROXIES;
 	autoResolverCount = CLIENT_KNOBS->DEFAULT_AUTO_RESOLVERS;
 	autoDesiredTLogCount = CLIENT_KNOBS->DEFAULT_AUTO_LOGS;
@@ -166,6 +167,7 @@ bool DatabaseConfiguration::isValid() const {
 		getDesiredLogs() >= 1 &&
 		getDesiredResolvers() >= 1 &&
 		tLogDataStoreType != KeyValueStoreType::END &&
+		tLogSpillType != TLogSpillType::UNSET &&
 		storageServerStoreType != KeyValueStoreType::END &&
 		autoMasterProxyCount >= 1 &&
 		autoResolverCount >= 1 &&
@@ -255,6 +257,10 @@ StatusObject DatabaseConfiguration::toJSON(bool noPolicies) const {
 			result["storage_engine"] = "custom";
 		}
 
+		if ( tLogSpillType != TLogSpillType::DEFAULT ) {
+			result["log_spill"] = (int)tLogSpillType;
+		}
+
 		if( remoteTLogReplicationFactor == 1 ) {
 			result["remote_redundancy_mode"] = "remote_single";
 		} else if( remoteTLogReplicationFactor == 2 ) {
@@ -371,6 +377,7 @@ bool DatabaseConfiguration::setInternal(KeyRef key, ValueRef value) {
 		if(tLogDataStoreType == KeyValueStoreType::SSD_REDWOOD_V1)
 			tLogDataStoreType = KeyValueStoreType::SSD_BTREE_V2;
 	}
+	else if (ck == LiteralStringRef("log_spill")) { parse((&type), value); tLogSpillType = (TLogSpillType::SpillType)type; }
 	else if (ck == LiteralStringRef("storage_engine")) { parse((&type), value); storageServerStoreType = (KeyValueStoreType::StoreType)type; }
 	else if (ck == LiteralStringRef("auto_proxies")) parse(&autoMasterProxyCount, value);
 	else if (ck == LiteralStringRef("auto_resolvers")) parse(&autoResolverCount, value);

fdbclient/DatabaseConfiguration.h

@@ -162,6 +162,7 @@ struct DatabaseConfiguration {
 	int32_t tLogWriteAntiQuorum;
 	int32_t tLogReplicationFactor;
 	KeyValueStoreType tLogDataStoreType;
+	TLogSpillType tLogSpillType;
 
 	// Storage Servers
 	IRepPolicyRef storagePolicy;

fdbclient/FDBTypes.h

@@ -513,6 +513,37 @@ private:
 	uint32_t type;
 };
 
+struct TLogSpillType {
+	// These enumerated values are stored in the database configuration, so can NEVER be changed.  Only add new ones just before END.
+	enum SpillType {
+		DEFAULT = 2,
+		VALUE = 1,
+		REFERENCE = 2,
+		UNSET,
+	};
+
+	TLogSpillType() : type(DEFAULT) {}
+	TLogSpillType( SpillType type ) : type(type) {
+		if ((uint32_t)type > UNSET)
+			this->type = UNSET;
+	}
+	operator SpillType() const { return SpillType(type); }
+
+	template <class Ar>
+	void serialize(Ar& ar) { serializer(ar, type); }
+
+	std::string toString() const {
+		switch( type ) {
+			case VALUE: return "";  // For backwards compatiblity.
+			case REFERENCE: return "reference";
+			default: return "unknown";
+		}
+	}
+
+private:
+	uint32_t type;
+};
+
 //Contains the amount of free and total space for a storage server, in bytes
 struct StorageBytes {
 	int64_t free;

fdbserver/TagPartitionedLogSystem.actor.cpp

@@ -1953,6 +1953,7 @@ struct TagPartitionedLogSystem : ILogSystem, ReferenceCounted<TagPartitionedLogS
 			InitializeTLogRequest &req = reqs[i];
 			req.recruitmentID = logSystem->recruitmentID;
 			req.storeType = configuration.tLogDataStoreType;
+			req.spillType = configuration.tLogSpillType;
 			req.recoverFrom = oldLogSystem->getLogSystemConfig();
 			req.recoverAt = oldLogSystem->recoverAt.get();
 			req.knownCommittedVersion = oldLogSystem->knownCommittedVersion;
@@ -1996,6 +1997,7 @@ struct TagPartitionedLogSystem : ILogSystem, ReferenceCounted<TagPartitionedLogS
 				InitializeTLogRequest &req = sreqs[i];
 				req.recruitmentID = logSystem->recruitmentID;
 				req.storeType = configuration.tLogDataStoreType;
+				req.spillType = configuration.tLogSpillType;
 				req.recoverFrom = oldLogSystem->getLogSystemConfig();
 				req.recoverAt = oldLogSystem->recoverAt.get();
 				req.knownCommittedVersion = oldLogSystem->knownCommittedVersion;