From 9e20b08976b49bbbe7216c55d757b7b4fb43577d Mon Sep 17 00:00:00 2001
From: Lukas Joswiak <lukas.joswiak@snowflake.com>
Date: Wed, 17 Feb 2021 16:04:23 -0800
Subject: [PATCH] Add float and double parsing

---
 fdbclient/GlobalConfig.actor.cpp |  9 ++++-
 fdbclient/Tuple.cpp              | 64 ++++++++++++++++++++++++++++++++
 fdbclient/Tuple.h                |  4 +-
 3 files changed, 74 insertions(+), 3 deletions(-)

diff --git a/fdbclient/GlobalConfig.actor.cpp b/fdbclient/GlobalConfig.actor.cpp
index f4c2c81e5d..8997d11973 100644
--- a/fdbclient/GlobalConfig.actor.cpp
+++ b/fdbclient/GlobalConfig.actor.cpp
@@ -40,7 +40,7 @@ GlobalConfig& GlobalConfig::globalConfig() {
 const std::any GlobalConfig::get(StringRef name) {
 	auto it = data.find(name);
 	if (it == data.end()) {
-		return nullptr;
+		return std::any{};
 	}
 	return it->second;
 }
@@ -51,10 +51,15 @@ Future<Void> GlobalConfig::onInitialized() {
 
 void GlobalConfig::insert(KeyRef key, ValueRef value) {
 	Tuple t = Tuple::unpack(value);
-	// TODO: Add more Tuple types
 	if (t.getType(0) == Tuple::ElementType::UTF8) {
 		data[key] = t.getString(0);
 	} else if (t.getType(0) == Tuple::ElementType::INT) {
 		data[key] = t.getInt(0);
+	} else if (t.getType(0) == Tuple::ElementType::FLOAT) {
+		data[key] = t.getFloat(0);
+	} else if (t.getType(0) == Tuple::ElementType::DOUBLE) {
+		data[key] = t.getDouble(0);
+	} else {
+		ASSERT(false);
 	}
 }
diff --git a/fdbclient/Tuple.cpp b/fdbclient/Tuple.cpp
index 3d4427079f..535be3d7fc 100644
--- a/fdbclient/Tuple.cpp
+++ b/fdbclient/Tuple.cpp
@@ -20,6 +20,18 @@
 
 #include "fdbclient/Tuple.h"
 
+static float bigEndianFloat(float orig) {
+	int32_t big = *(int32_t*)&orig;
+	big = bigEndian32(big);
+	return *(float*)&big;
+}
+
+static double bigEndianDouble(double orig) {
+	int64_t big = *(int64_t*)&orig;
+	big = bigEndian64(big);
+	return *(double*)&big;
+}
+
 static size_t find_string_terminator(const StringRef data, size_t offset) {
 	size_t i = offset;
 	while (i < data.size() - 1 && !(data[i] == '\x00' && data[i + 1] != (uint8_t)'\xff')) {
@@ -29,6 +41,19 @@ static size_t find_string_terminator(const StringRef data, size_t offset) {
 	return i;
 }
 
+// If encoding and the sign bit is 1 (the number is negative), flip all the bits.
+// If decoding and the sign bit is 0 (the number is negative), flip all the bits.
+// Otherwise, the number is positive, so flip the sign bit.
+static void adjust_floating_point(uint8_t *bytes, size_t size, bool encode) {
+	if((encode && ((uint8_t)(bytes[0] & 0x80) != (uint8_t)0x00)) || (!encode && ((uint8_t)(bytes[0] & 0x80) != (uint8_t)0x80))) {
+		for(size_t i = 0; i < size; i++) {
+			bytes[i] ^= (uint8_t)0xff;
+		}
+	} else {
+		bytes[0] ^= (uint8_t)0x80;
+	}
+}
+
 Tuple::Tuple(StringRef const& str, bool exclude_incomplete) {
 	data.append(data.arena(), str.begin(), str.size());
 
@@ -228,6 +253,45 @@ int64_t Tuple::getInt(size_t index, bool allow_incomplete) const {
 	return swap;
 }
 
+// TODO: Combine with bindings/flow/Tuple.*. This code is copied from there.
+float Tuple::getFloat(size_t index) const {
+	if(index >= offsets.size()) {
+		throw invalid_tuple_index();
+	}
+	ASSERT_LT(offsets[index], data.size());
+	uint8_t code = data[offsets[index]];
+	if(code != 0x20) {
+		throw invalid_tuple_data_type();
+	}
+
+	float swap;
+	uint8_t* bytes = (uint8_t*)&swap;
+	ASSERT_LE(offsets[index] + 1 + sizeof(float), data.size());
+	swap = *(float*)(data.begin() + offsets[index] + 1);
+	adjust_floating_point( bytes, sizeof(float), false );
+
+	return bigEndianFloat(swap);
+}
+
+double Tuple::getDouble(size_t index) const {
+	if(index >= offsets.size()) {
+		throw invalid_tuple_index();
+	}
+	ASSERT_LT(offsets[index], data.size());
+	uint8_t code = data[offsets[index]];
+	if(code != 0x21) {
+		throw invalid_tuple_data_type();
+	}
+
+	double swap;
+	uint8_t* bytes = (uint8_t*)&swap;
+	ASSERT_LE(offsets[index] + 1 + sizeof(double), data.size());
+	swap = *(double*)(data.begin() + offsets[index] + 1);
+	adjust_floating_point( bytes, sizeof(double), false );
+
+	return bigEndianDouble(swap);
+}
+
 KeyRange Tuple::range(Tuple const& tuple) const {
 	VectorRef<uint8_t> begin;
 	VectorRef<uint8_t> end;
diff --git a/fdbclient/Tuple.h b/fdbclient/Tuple.h
index b44edd73cc..4497f19441 100644
--- a/fdbclient/Tuple.h
+++ b/fdbclient/Tuple.h
@@ -47,7 +47,7 @@ struct Tuple {
 		return append(t);
 	}
 
-	enum ElementType { NULL_TYPE, INT, BYTES, UTF8 };
+	enum ElementType { NULL_TYPE, INT, BYTES, UTF8, FLOAT, DOUBLE };
 
 	// this is number of elements, not length of data
 	size_t size() const { return offsets.size(); }
@@ -55,6 +55,8 @@ struct Tuple {
 	ElementType getType(size_t index) const;
 	Standalone<StringRef> getString(size_t index) const;
 	int64_t getInt(size_t index, bool allow_incomplete = false) const;
+	float getFloat(size_t index) const;
+	double getDouble(size_t index) const;
 
 	KeyRange range(Tuple const& tuple = Tuple()) const;