Native precise indexing of decfloat fields

src/common/DecFloat.cpp

@@ -106,12 +106,8 @@ private:
 
 CDecimal128 dmax(DBL_MAX, DecimalStatus(0)), dmin(-DBL_MAX, DecimalStatus(0));
 
-void make(unsigned int* key,
-	const unsigned pMax, const int bias, const unsigned decSize,
-	unsigned char* coeff, int sign, int exp)
+unsigned digits(const unsigned pMax, unsigned char* const coeff, int& exp)
 {
-	// normalize coeff & exponent
-	bool zero = true;
 	for (unsigned i = 0; i < pMax; ++i)
 	{
 		if (coeff[i])
@@ -122,14 +118,30 @@ void make(unsigned int* key,
 				memset(&coeff[pMax - i], 0, i);
 				exp -= i;
 			}
-			zero = false;
-			break;
+
+			i = pMax - i;
+			while (!coeff[i - 1])
+			{
+				fb_assert(i > 0);
+				--i;
+			}
+			return i;
 		}
 	}
 
+	return 0;
+}
+
+void make(unsigned int* key,
+	const unsigned pMax, const int bias, const unsigned decSize,
+	unsigned char* coeff, int sign, int exp)
+{
+	// normalize coeff & exponent
+	unsigned dig = digits(pMax, coeff, exp);
+
 	// exponent bias and sign
 	exp += (bias + 2);
-	if (zero)
+	if (!dig)
 		exp = 1;
 	if (sign)
 		exp = -exp;
@@ -146,6 +158,75 @@ void make(unsigned int* key,
 	}
 }
 
+struct tab { UCHAR rshift, lshift; };
+tab table[4] =
+{
+	{ 2, 6 },
+	{ 4, 4 },
+	{ 6, 2 },
+	{ 8, 0 }
+};
+
+ULONG indexKey(char* key, unsigned pMax, const int bias, unsigned char* coeff, int sign, int exp)
+{
+	// normalize coeff & exponent
+	unsigned dig = digits(pMax, coeff, exp);
+
+	// exponent bias and sign
+	exp += (bias + 1);
+	if (!dig)
+		exp = 0;
+	if (sign)
+		exp = -exp;
+	exp += 2 * (bias + 1);	// make it positive
+	fb_assert(exp >= 0 && exp < 64 * 1024);
+
+	// encode exp
+	char* k = key;
+	*k++ = exp >> 8;
+	*k++ = exp & 0xff;
+
+	// invert negative
+	unsigned char* const end = &coeff[dig];
+	if (sign && dig)
+	{
+		fb_assert(end[-1]);
+		--end[-1];
+
+		for (unsigned char* p = coeff; p < end; ++p)
+			*p = 9 - *p;
+	}
+
+	// Some 0's in the end - caller, do not forget to reserve additional space on stack
+	end[0] = end[1] = 0;
+
+	// Avoid bad data in k in case when coeff is zero
+	*k = 0;
+
+	// compress coeff - 3 decimal digits (999) per 10 bits (1023)
+	unsigned char* p = coeff;
+	for (tab* t = table; p < end; p += 3)
+	{
+		USHORT val = p[0] * 100 + p[1] * 10 + p[2];
+		fb_assert(val < 1000);	// 1024, 10 bit
+		*k |= (val >> t->rshift);
+		++k;
+		*k = (val << t->lshift);
+		if (!t->lshift)
+		{
+			++k;
+			*k = 0;
+			t = table;
+		}
+		else
+			++t;
+	}
+	if (*k)
+		++k;
+
+	return k - key;
+}
+
 void grab(unsigned int* key,
 	const unsigned pMax, const int bias, const unsigned decSize,
 	unsigned char* bcd, int& sign, int& exp)
@@ -346,6 +427,39 @@ Decimal64 Decimal64::neg() const
 	return rc;
 }
 
+void Decimal64::makeKey(unsigned int* key) const
+{
+	unsigned char coeff[DECDOUBLE_Pmax];
+	int sign = decDoubleGetCoefficient(&dec, coeff);
+	int exp = decDoubleGetExponent(&dec);
+
+	make(key, DECDOUBLE_Pmax, DECDOUBLE_Bias, sizeof(dec), coeff, sign, exp);
+}
+
+void Decimal64::grabKey(unsigned int* key)
+{
+	int exp, sign;
+	unsigned char bcd[DECDOUBLE_Pmax];
+
+	grab(key, DECDOUBLE_Pmax, DECDOUBLE_Bias, sizeof(dec), bcd, sign, exp);
+
+	decDoubleFromBCD(&dec, exp, bcd, sign);
+}
+
+ULONG Decimal64::getIndexKeyLength()
+{
+	return 9;
+}
+
+ULONG Decimal64::makeIndexKey(vary* buf)
+{
+	unsigned char coeff[DECDOUBLE_Pmax + 2];
+	int sign = decDoubleGetCoefficient(&dec, coeff);
+	int exp = decDoubleGetExponent(&dec);
+
+	buf->vary_length = indexKey(buf->vary_string, DECDOUBLE_Pmax, DECDOUBLE_Bias, coeff, sign, exp);
+	return buf->vary_length;
+}
 
 
 Decimal128 Decimal128::set(Decimal64 d64)
@@ -678,23 +792,19 @@ void Decimal128::grabKey(unsigned int* key)
 	decQuadFromBCD(&dec, exp, bcd, sign);
 }
 
-void Decimal64::makeKey(unsigned int* key) const
+ULONG Decimal128::getIndexKeyLength()
 {
-	unsigned char coeff[DECDOUBLE_Pmax];
-	int sign = decDoubleGetCoefficient(&dec, coeff);
-	int exp = decDoubleGetExponent(&dec);
-
-	make(key, DECDOUBLE_Pmax, DECDOUBLE_Bias, sizeof(dec), coeff, sign, exp);
+	return 17;
 }
 
-void Decimal64::grabKey(unsigned int* key)
+ULONG Decimal128::makeIndexKey(vary* buf)
 {
-	int exp, sign;
-	unsigned char bcd[DECDOUBLE_Pmax];
+	unsigned char coeff[DECQUAD_Pmax + 2];
+	int sign = decQuadGetCoefficient(&dec, coeff);
+	int exp = decQuadGetExponent(&dec);
 
-	grab(key, DECDOUBLE_Pmax, DECDOUBLE_Bias, sizeof(dec), bcd, sign, exp);
-
-	decDoubleFromBCD(&dec, exp, bcd, sign);
+	buf->vary_length = indexKey(buf->vary_string, DECQUAD_Pmax, DECQUAD_Bias, coeff, sign, exp);
+	return buf->vary_length;
 }
 
 } // namespace Firebird

src/common/DecFloat.h

@@ -76,6 +76,8 @@ public:
 
 	void makeKey(unsigned int* key) const;
 	void grabKey(unsigned int* key);
+	static ULONG getIndexKeyLength();
+	ULONG makeIndexKey(vary* buf);
 
 #ifdef DEV_BUILD
 	int show();
@@ -127,6 +129,8 @@ public:
 
 	void makeKey(unsigned int* key) const;
 	void grabKey(unsigned int* key);
+	static ULONG getIndexKeyLength();
+	ULONG makeIndexKey(vary* buf);
 
 #ifdef DEV_BUILD
 	int show();

src/jrd/btr.cpp

@@ -1373,6 +1373,14 @@ USHORT BTR_key_length(thread_db* tdbb, jrd_rel* relation, index_desc* idx)
 			length = sizeof(UCHAR);
 			break;
 
+		case dtype_dec64:
+			length = Decimal64::getIndexKeyLength();
+			break;
+
+		case dtype_dec128:
+			length = Decimal128::getIndexKeyLength();
+			break;
+
 		default:
 			if (idx->idx_flags & idx_expressn)
 			{
@@ -1425,6 +1433,12 @@ USHORT BTR_key_length(thread_db* tdbb, jrd_rel* relation, index_desc* idx)
 		case idx_boolean:
 			length = sizeof(UCHAR);
 			break;
+		case dtype_dec64:
+			length = Decimal64::getIndexKeyLength();
+			break;
+		case dtype_dec128:
+			length = Decimal128::getIndexKeyLength();
+			break;
 		default:
 			length = format->fmt_desc[tail->idx_field].dsc_length;
 			if (format->fmt_desc[tail->idx_field].dsc_dtype == dtype_varying)
@@ -2423,6 +2437,7 @@ static void compress(thread_db* tdbb,
 	}
 
 	if (itype == idx_string || itype == idx_byte_array || itype == idx_metadata ||
+		itype == idx_dec64 || itype == idx_dec128 ||
 		itype >= idx_first_intl_string)
 	{
 		VaryStr<MAX_KEY> buffer;
@@ -2431,7 +2446,19 @@ static void compress(thread_db* tdbb,
 
 		size_t length;
 
-		if (itype >= idx_first_intl_string || itype == idx_metadata)
+		if (itype == idx_dec64)
+		{
+			Decimal64 dec = MOV_get_dec64(tdbb, desc);
+			length = dec.makeIndexKey(&buffer);
+			ptr = reinterpret_cast<UCHAR*>(buffer.vary_string);
+		}
+		else if (itype == idx_dec128)
+		{
+			Decimal128 dec = MOV_get_dec128(tdbb, desc);
+			length = dec.makeIndexKey(&buffer);
+			ptr = reinterpret_cast<UCHAR*>(buffer.vary_string);
+		}
+		else if (itype >= idx_first_intl_string || itype == idx_metadata)
 		{
 			DSC to;
 

src/jrd/btr.h

@@ -99,6 +99,8 @@ const int idx_sql_time		= 6;
 const int idx_timestamp		= 7;
 const int idx_numeric2		= 8;	// Introduced for 64-bit Integer support
 const int idx_boolean		= 9;
+const int idx_dec64			= 10;
+const int idx_dec128		= 11;
 
 				   // idx_itype space for future expansion
 const int idx_first_intl_string	= 64;	// .. MAX (short) Range of computed key strings

src/jrd/dfw.epp

@@ -1241,6 +1241,10 @@ USHORT DFW_assign_index_type(thread_db* tdbb, const Firebird::MetaName& name, SS
 		return idx_numeric2;
 	case dtype_boolean:
 		return idx_boolean;
+	case dtype_dec64:
+		return idx_dec64;
+	case dtype_dec128:
+		return idx_dec128;
 	default:
 		return idx_numeric;
 	}