UniformRangePartition extends RangePartitionAlgorithm { public Tuple increment(final Tuple last, BigInteger interval, final int baseDigit) { BigInteger [] incs = new BigInteger[last.size()]; boolean [] overflowFlag = new boolean[last.size()]; BigInteger [] result; BigInteger value = interval; BigInteger [] reverseCardsForDigit = new BigInteger[baseDigit + 1]; for (int i = baseDigit; i >= 0; i--) { if (i == baseDigit) { reverseCardsForDigit[i] = colCards[i]; } else { reverseCardsForDigit[i] = reverseCardsForDigit[i+1].multiply(colCards[i]); } } for (int i = 0; i < baseDigit; i++) { result = value.divideAndRemainder(reverseCardsForDigit[i + 1]); incs[i] = result[0]; value = result[1]; } int finalId = baseDigit; incs[finalId] = value; for (int i = finalId; i >= 0; i--) { if (isOverflow(i, last.get(i), incs[i], sortSpecs)) { if (i == 0) { throw new RangeOverflowException(mergedRange, last, incs[i].longValue(), sortSpecs[i].isAscending()); } long rem = incrementAndGetReminder(i, last.get(i), value.longValue()); incs[i] = BigInteger.valueOf(rem); incs[i - 1] = incs[i-1].add(BigInteger.ONE); overflowFlag[i] = true; } else { if (i > 0) { incs[i] = value; break; } } } for (int i = 0; i < incs.length; i++) { if (incs[i] == null) { incs[i] = BigInteger.ZERO; } } Tuple end = new VTuple(sortSpecs.length); Column column; for (int i = 0; i < last.size(); i++) { column = sortSpecs[i].getSortKey(); switch (column.getDataType().getType()) { case CHAR: if (overflowFlag[i]) { end.put(i, DatumFactory.createChar((char) (mergedRange.getStart().get(i).asChar() + incs[i].longValue()))); } else { if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createChar((char) (last.get(i).asChar() + incs[i].longValue()))); } else { end.put(i, DatumFactory.createChar((char) (last.get(i).asChar() - incs[i].longValue()))); } } break; case BIT: if (overflowFlag[i]) { end.put(i, DatumFactory.createBit( (byte) (mergedRange.getStart().get(i).asByte() + incs[i].longValue()))); } else { if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createBit((byte) (last.get(i).asByte() + incs[i].longValue()))); } else { end.put(i, DatumFactory.createBit((byte) (last.get(i).asByte() - incs[i].longValue()))); } } break; case INT2: if (overflowFlag[i]) { end.put(i, DatumFactory.createInt2( (short) (mergedRange.getStart().get(i).asInt2() + incs[i].longValue()))); } else { if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createInt2((short) (last.get(i).asInt2() + incs[i].longValue()))); } else { end.put(i, DatumFactory.createInt2((short) (last.get(i).asInt2() - incs[i].longValue()))); } } break; case INT4: if (overflowFlag[i]) { end.put(i, DatumFactory.createInt4( (int) (mergedRange.getStart().get(i).asInt4() + incs[i].longValue()))); } else { if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createInt4((int) (last.get(i).asInt4() + incs[i].longValue()))); } else { end.put(i, DatumFactory.createInt4((int) (last.get(i).asInt4() - incs[i].longValue()))); } } break; case INT8: if (overflowFlag[i]) { end.put(i, DatumFactory.createInt8( mergedRange.getStart().get(i).asInt8() + incs[i].longValue())); } else { if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createInt8(last.get(i).asInt8() + incs[i].longValue())); } else { end.put(i, DatumFactory.createInt8(last.get(i).asInt8() - incs[i].longValue())); } } break; case FLOAT4: if (overflowFlag[i]) { end.put(i, DatumFactory.createFloat4( mergedRange.getStart().get(i).asFloat4() + incs[i].longValue())); } else { if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createFloat4(last.get(i).asFloat4() + incs[i].longValue())); } else { end.put(i, DatumFactory.createFloat4(last.get(i).asFloat4() - incs[i].longValue())); } } break; case FLOAT8: if (overflowFlag[i]) { end.put(i, DatumFactory.createFloat8( mergedRange.getStart().get(i).asFloat8() + incs[i].longValue())); } else { if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createFloat8(last.get(i).asFloat8() + incs[i].longValue())); } else { end.put(i, DatumFactory.createFloat8(last.get(i).asFloat8() - incs[i].longValue())); } } break; case TEXT: if (overflowFlag[i]) { end.put(i, DatumFactory.createText(((char) (mergedRange.getStart().get(i).asChars().charAt(0) + incs[i].longValue())) + "")); } else { BigInteger lastBigInt; if (last.isNull(i)) { lastBigInt = BigInteger.valueOf(0); end.put(i, DatumFactory.createText(lastBigInt.add(incs[i]).toByteArray())); } else { if (isPureAscii[i]) { lastBigInt = new BigInteger(last.get(i).asByteArray()); if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createText(lastBigInt.add(incs[i]).toByteArray())); } else { end.put(i, DatumFactory.createText(lastBigInt.subtract(incs[i]).toByteArray())); } } else { char[] lastChars = last.getUnicodeChars(i); int [] charIncs = new int[lastChars.length]; BigInteger remain = incs[i]; for (int k = lastChars.length - 1; k > 0 && remain.compareTo(BigInteger.ZERO) > 0; k--) { BigInteger digitBase = BigInteger.valueOf(TextDatum.UNICODE_CHAR_BITS_NUM).pow(k); if (remain.compareTo(digitBase) > 0) { charIncs[k] = remain.divide(digitBase).intValue(); BigInteger sub = digitBase.multiply(BigInteger.valueOf(charIncs[k])); remain = remain.subtract(sub); } } charIncs[charIncs.length - 1] = remain.intValue(); for (int k = 0; k < lastChars.length; k++) { if (charIncs[k] == 0) { continue; } if (sortSpecs[i].isAscending()) { int sum = (int) lastChars[k] + charIncs[k]; if (sum > TextDatum.UNICODE_CHAR_BITS_NUM) { charIncs[k] = sum - TextDatum.UNICODE_CHAR_BITS_NUM; charIncs[k - 1] += 1; lastChars[k - 1] += 1; lastChars[k] += charIncs[k]; } else { lastChars[k] += charIncs[k]; } } else { int sum = (int) lastChars[k] - charIncs[k]; if (sum < 0) { charIncs[k] = TextDatum.UNICODE_CHAR_BITS_NUM - sum; charIncs[k - 1] -= 1; lastChars[k - 1] -= 1; lastChars[k] += charIncs[k]; } else { lastChars[k] -= charIncs[k]; } } } end.put(i, DatumFactory.createText(Convert.chars2utf(lastChars))); } } } break; case DATE: if (overflowFlag[i]) { end.put(i, DatumFactory.createDate((int) (mergedRange.getStart().get(i).asInt4() + incs[i].longValue()))); } else { if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createDate((int) (last.get(i).asInt4() + incs[i].longValue()))); } else { end.put(i, DatumFactory.createDate((int) (last.get(i).asInt4() - incs[i].longValue()))); } } break; case TIME: if (overflowFlag[i]) { end.put(i, DatumFactory.createTime(mergedRange.getStart().get(i).asInt8() + incs[i].longValue())); } else { if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createTime(last.get(i).asInt8() + incs[i].longValue())); } else { end.put(i, DatumFactory.createTime(last.get(i).asInt8() - incs[i].longValue())); } } break; case TIMESTAMP: if (overflowFlag[i]) { end.put(i, DatumFactory.createTimestmpDatumWithJavaMillis( mergedRange.getStart().get(i).asInt8() + incs[i].longValue())); } else { if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createTimestmpDatumWithJavaMillis(last.get(i).asInt8() + incs[i].longValue())); } else { end.put(i, DatumFactory.createTimestmpDatumWithJavaMillis(last.get(i).asInt8() - incs[i].longValue())); } } break; case INET4: byte[] ipBytes; if (overflowFlag[i]) { ipBytes = mergedRange.getStart().get(i).asByteArray(); assert ipBytes.length == 4; end.put(i, DatumFactory.createInet4(ipBytes)); } else { if (sortSpecs[i].isAscending()) { int lastVal = last.get(i).asInt4() + incs[i].intValue(); ipBytes = new byte[4]; Bytes.putInt(ipBytes, 0, lastVal); end.put(i, DatumFactory.createInet4(ipBytes)); } else { int lastVal = last.get(i).asInt4() - incs[i].intValue(); ipBytes = new byte[4]; Bytes.putInt(ipBytes, 0, lastVal); end.put(i, DatumFactory.createInet4(ipBytes)); } } break; default: throw new UnsupportedOperationException(column.getDataType() + " is not supported yet"); } if (beginNulls[i] && endNulls[i]) { end.put(i, NullDatum.get()); continue; } } return end; } UniformRangePartition(final TupleRange entireRange, final SortSpec[] sortSpecs, boolean inclusive);  UniformRangePartition(TupleRange range, SortSpec [] sortSpecs); @Override TupleRange[] partition(int partNum); void normalize(final SortSpec [] sortSpecs, TupleRange range); boolean isOverflow(int colId, Datum last, BigInteger inc, SortSpec [] sortSpecs); long incrementAndGetReminder(int colId, Datum last, long inc); Tuple increment(final Tuple last, BigInteger interval, final int baseDigit); static BigInteger charsToBigInteger(char [] chars);  }
@Test public void testIncrementOfText() { Schema schema = new Schema() .addColumn("l_returnflag", Type.TEXT) .addColumn("l_linestatus", Type.TEXT); SortSpec[] sortSpecs = PlannerUtil.schemaToSortSpecs(schema); Tuple s = new VTuple(2); s.put(0, DatumFactory.createText("A")); s.put(1, DatumFactory.createText("A")); Tuple e = new VTuple(2); e.put(0, DatumFactory.createText("D")); e.put(1, DatumFactory.createText("C")); TupleRange expected = new TupleRange(sortSpecs, s, e); UniformRangePartition partitioner = new UniformRangePartition(expected, sortSpecs); assertEquals(12, partitioner.getTotalCardinality().intValue()); String [] result = new String[12]; result[0] = "AA"; result[1] = "AB"; result[2] = "AC"; result[3] = "BA"; result[4] = "BB"; result[5] = "BC"; result[6] = "CA"; result[7] = "CB"; result[8] = "CC"; result[9] = "DA"; result[10] = "DB"; result[11] = "DC"; Tuple end = partitioner.increment(s, BigInteger.valueOf(1), 1); assertEquals("A", end.get(0).asChars()); assertEquals("B", end.get(1).asChars()); for (int i = 2; i < 11; i++ ) { end = partitioner.increment(end, BigInteger.valueOf(1), 1); assertEquals(result[i].charAt(0), end.get(0).asChars().charAt(0)); assertEquals(result[i].charAt(1), end.get(1).asChars().charAt(0)); } }
@Test public void testIncrementOfText2() { Schema schema = new Schema() .addColumn("l_returnflag", Type.TEXT) .addColumn("l_linestatus", Type.TEXT); SortSpec [] sortSpecs = PlannerUtil.schemaToSortSpecs(schema); Tuple s = new VTuple(2); s.put(0, DatumFactory.createText("A")); s.put(1, DatumFactory.createText("A")); Tuple e = new VTuple(2); e.put(0, DatumFactory.createText("D")); e.put(1, DatumFactory.createText("C")); TupleRange expected = new TupleRange(sortSpecs, s, e); UniformRangePartition partitioner = new UniformRangePartition(expected, sortSpecs); assertEquals(12, partitioner.getTotalCardinality().intValue()); String [] result = new String[12]; result[0] = "AA"; result[1] = "AB"; result[2] = "AC"; result[3] = "BA"; result[4] = "BB"; result[5] = "BC"; result[6] = "CA"; result[7] = "CB"; result[8] = "CC"; result[9] = "DA"; result[10] = "DB"; result[11] = "DC"; Tuple end = partitioner.increment(s, BigInteger.valueOf(6), 1); assertEquals("C", end.get(0).asChars()); assertEquals("A", end.get(1).asChars()); end = partitioner.increment(end, BigInteger.valueOf(5), 1); assertEquals("D", end.get(0).asChars()); assertEquals("C", end.get(1).asChars()); }
@Test public void testIncrementOfText3() { Schema schema = new Schema() .addColumn("l_returnflag", Type.TEXT) .addColumn("l_linestatus", Type.TEXT) .addColumn("final", Type.TEXT); SortSpec [] sortSpecs = PlannerUtil.schemaToSortSpecs(schema); Tuple s = new VTuple(3); s.put(0, DatumFactory.createText("A")); s.put(1, DatumFactory.createText("A")); s.put(2, DatumFactory.createText("A")); Tuple e = new VTuple(3); e.put(0, DatumFactory.createText("D")); e.put(1, DatumFactory.createText("B")); e.put(2, DatumFactory.createText("C")); TupleRange expected = new TupleRange(sortSpecs, s, e); UniformRangePartition partitioner = new UniformRangePartition(expected, sortSpecs); assertEquals(24, partitioner.getTotalCardinality().intValue()); Tuple overflowBefore = partitioner.increment(s, BigInteger.valueOf(5), 2); assertEquals("A", overflowBefore.get(0).asChars()); assertEquals("B", overflowBefore.get(1).asChars()); assertEquals("C", overflowBefore.get(2).asChars()); Tuple overflowed = partitioner.increment(overflowBefore, BigInteger.valueOf(1), 2); assertEquals("B", overflowed.get(0).asChars()); assertEquals("A", overflowed.get(1).asChars()); assertEquals("A", overflowed.get(2).asChars()); }
@Test public void testIncrementOfUnicode() { Schema schema = new Schema() .addColumn("col1", Type.TEXT); SortSpec [] sortSpecs = PlannerUtil.schemaToSortSpecs(schema); Tuple s = new VTuple(1); s.put(0, DatumFactory.createText("가가가")); Tuple e = new VTuple(1); e.put(0, DatumFactory.createText("하하하")); TupleRange expected = new TupleRange(sortSpecs, s, e); UniformRangePartition partitioner = new UniformRangePartition(expected, sortSpecs); BaseTupleComparator comp = new BaseTupleComparator(schema, sortSpecs); Tuple tuple = s; Tuple prevTuple = null; for (int i = 0; i < 100; i++) { tuple = partitioner.increment(tuple, BigInteger.valueOf(30000), 0); if (prevTuple != null) { assertTrue("prev=" + prevTuple + ", current=" + tuple, comp.compare(prevTuple, tuple) < 0); } prevTuple = tuple; } }
@Test public void testIncrementOfInt8() { Schema schema = new Schema() .addColumn("l_orderkey", Type.INT8) .addColumn("l_linenumber", Type.INT8); SortSpec [] sortSpecs = PlannerUtil.schemaToSortSpecs(schema); Tuple s = new VTuple(2); s.put(0, DatumFactory.createInt8(10)); s.put(1, DatumFactory.createInt8(20)); Tuple e = new VTuple(2); e.put(0, DatumFactory.createInt8(19)); e.put(1, DatumFactory.createInt8(39)); TupleRange expected = new TupleRange(sortSpecs, s, e); UniformRangePartition partitioner = new UniformRangePartition(expected, sortSpecs); assertEquals(200, partitioner.getTotalCardinality().longValue()); Tuple range2 = partitioner.increment(s, BigInteger.valueOf(100), 1); assertEquals(15, range2.get(0).asInt4()); assertEquals(20, range2.get(1).asInt4()); Tuple range3 = partitioner.increment(range2, BigInteger.valueOf(99), 1); assertEquals(19, range3.get(0).asInt4()); assertEquals(39, range3.get(1).asInt4()); }
@Test public void testIncrementOfInt8AndFinal() { Schema schema = new Schema() .addColumn("l_orderkey", Type.INT8) .addColumn("l_linenumber", Type.INT8) .addColumn("final", Type.INT8); SortSpec [] sortSpecs = PlannerUtil.schemaToSortSpecs(schema); Tuple s = new VTuple(3); s.put(0, DatumFactory.createInt8(1)); s.put(1, DatumFactory.createInt8(1)); s.put(2, DatumFactory.createInt8(1)); Tuple e = new VTuple(3); e.put(0, DatumFactory.createInt8(4)); e.put(1, DatumFactory.createInt8(2)); e.put(2, DatumFactory.createInt8(3)); TupleRange expected = new TupleRange(sortSpecs, s, e); UniformRangePartition partitioner = new UniformRangePartition(expected, sortSpecs); assertEquals(24, partitioner.getTotalCardinality().longValue()); Tuple beforeOverflow = partitioner.increment(s, BigInteger.valueOf(5), 2); assertEquals(1, beforeOverflow.get(0).asInt8()); assertEquals(2, beforeOverflow.get(1).asInt8()); assertEquals(3, beforeOverflow.get(2).asInt8()); Tuple overflow = partitioner.increment(beforeOverflow, BigInteger.valueOf(1), 2); assertEquals(2, overflow.get(0).asInt8()); assertEquals(1, overflow.get(1).asInt8()); assertEquals(1, overflow.get(2).asInt8()); }
@Test public void testIncrementOfFloat8() { Schema schema = new Schema() .addColumn("l_orderkey", Type.FLOAT8) .addColumn("l_linenumber", Type.FLOAT8) .addColumn("final", Type.FLOAT8); SortSpec [] sortSpecs = PlannerUtil.schemaToSortSpecs(schema); Tuple s = new VTuple(3); s.put(0, DatumFactory.createFloat8(1.1d)); s.put(1, DatumFactory.createFloat8(1.1d)); s.put(2, DatumFactory.createFloat8(1.1d)); Tuple e = new VTuple(3); e.put(0, DatumFactory.createFloat8(4.1d)); e.put(1, DatumFactory.createFloat8(2.1d)); e.put(2, DatumFactory.createFloat8(3.1d)); TupleRange expected = new TupleRange(sortSpecs, s, e); UniformRangePartition partitioner = new UniformRangePartition(expected, sortSpecs); assertEquals(24, partitioner.getTotalCardinality().longValue()); Tuple beforeOverflow = partitioner.increment(s, BigInteger.valueOf(5), 2); assertTrue(1.1d == beforeOverflow.get(0).asFloat8()); assertTrue(2.1d == beforeOverflow.get(1).asFloat8()); assertTrue(3.1d == beforeOverflow.get(2).asFloat8()); Tuple overflow = partitioner.increment(beforeOverflow, BigInteger.valueOf(1), 2); assertTrue(2.1d == overflow.get(0).asFloat8()); assertTrue(1.1d == overflow.get(1).asFloat8()); assertTrue(1.1d == overflow.get(2).asFloat8()); }
@Test public void testIncrementOfInet4() { Schema schema = new Schema() .addColumn("l_orderkey", Type.INET4) .addColumn("l_linenumber", Type.INET4) .addColumn("final", Type.INET4); SortSpec [] sortSpecs = PlannerUtil.schemaToSortSpecs(schema); Tuple s = new VTuple(3); s.put(0, DatumFactory.createInet4("127.0.1.1")); s.put(1, DatumFactory.createInet4("127.0.0.1")); s.put(2, DatumFactory.createInet4("128.0.0.253")); Tuple e = new VTuple(3); e.put(0, DatumFactory.createInet4("127.0.1.4")); e.put(1, DatumFactory.createInet4("127.0.0.2")); e.put(2, DatumFactory.createInet4("128.0.0.255")); TupleRange expected = new TupleRange(sortSpecs, s, e); UniformRangePartition partitioner = new UniformRangePartition(expected, sortSpecs); assertEquals(24, partitioner.getTotalCardinality().longValue()); Tuple beforeOverflow = partitioner.increment(s, BigInteger.valueOf(5), 2); assertTrue("127.0.1.1".equals(beforeOverflow.get(0).asChars())); assertTrue("127.0.0.2".equals(beforeOverflow.get(1).asChars())); assertTrue("128.0.0.255".equals(beforeOverflow.get(2).asChars())); Tuple overflow = partitioner.increment(beforeOverflow, BigInteger.valueOf(1), 2); assertTrue("127.0.1.2".equals(overflow.get(0).asChars())); assertTrue("127.0.0.1".equals(overflow.get(1).asChars())); assertTrue("128.0.0.253".equals(overflow.get(2).asChars())); }